from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("We don't swallow the internal error here, we raise again 😎")
        raise


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):

from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated


async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: Annotated[str, Header()]):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: set[str] = set()


@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name

from typing import Annotated

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Annotated[str | None, Query(min_length=3)] = ...):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})

from typing import Annotated

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/files/")
async def create_file(file: Annotated[bytes, File()]):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):

 * possibly run in another thread.
 *
 * <p>{@code UncheckedExecutionException} is intended as an alternative to {@code
 * ExecutionException} when the exception thrown by a task is an unchecked exception. However, it
 * may also wrap a checked exception in some cases.
 *
 * <p>When wrapping an {@code Error} from another thread, prefer {@link ExecutionError}. When

              "util.concurrent",
              "xml")
          .transform("com.google.common."::concat)
          .toSet();

  public void testClassesHaveOverrides() throws Exception {
    for (ClassInfo info : ClassPath.from(getClass().getClassLoader()).getAllClasses()) {
      if (!GUAVA_PACKAGES.contains(info.getPackageName())) {
        continue;
      }
      if (info.getName().endsWith("GwtSerializationDependencies")) {

        with:
          #Comma separated list of labels that can be assigned to issues to exclude them from being marked as stale
          exempt-issue-labels: 'override-stale'
          #Comma separated list of labels that can be assigned to PRs to exclude them from being marked as stale
          exempt-pr-labels: "override-stale"
          #Limit the No. of API calls in one run default value is 30.

import pytest
from dirty_equals import IsDict, IsOneOf
from fastapi.testclient import TestClient

from docs_src.response_model.tutorial004 import app

client = TestClient(app)


@pytest.mark.parametrize(
    "url,data",
    [
        ("/items/foo", {"name": "Foo", "price": 50.2}),
        (
            "/items/bar",
            {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
        ),
        (

import pytest
from fastapi.testclient import TestClient

from ...utils import needs_py310, needs_pydanticv2


@pytest.fixture(name="client")
def get_client() -> TestClient:
    from docs_src.separate_openapi_schemas.tutorial002_py310 import app

    client = TestClient(app)
    return client


@needs_py310
def test_create_item(client: TestClient) -> None:
    response = client.post("/items/", json={"name": "Foo"})