app = FastAPI()


@app.get("/catching")
def catching(d: Annotated[str, Depends(catching_dep)]) -> Any:
    raise CustomError("Simulated error during streaming")


@app.get("/broken")
def broken(d: Annotated[str, Depends(broken_dep)]) -> Any:
    return {"message": "all good?"}


client = TestClient(app)


def test_catching():
    response = client.get("/catching")
    assert response.status_code == 418

@app.put("/invalid-user/{user_id}")
def put_invalid_user(
    user_id: str, name: str = Body(), db: dict = Depends(get_database)
):
    db[user_id] = name
    raise HTTPException(status_code=400, detail="Invalid user")


@app.put("/user/{user_id}")
def put_user(user_id: str, name: str = Body(), db: dict = Depends(get_database)):
    db[user_id] = name
    return {"message": "OK"}


@pytest.fixture(autouse=True)

from collections.abc import Awaitable
from contextvars import ContextVar
from typing import Any, Callable, Optional

from fastapi import Depends, FastAPI, Request, Response
from fastapi.testclient import TestClient

legacy_request_state_context_var: ContextVar[Optional[dict[str, Any]]] = ContextVar(
    "legacy_request_state_context_var", default=None
)

app = FastAPI()


async def set_up_request_state_dependency():

from fastapi import Depends, FastAPI, Security
from fastapi.security import APIKeyCookie
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

api_key = APIKeyCookie(name="key")


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(api_key)):
    user = User(username=oauth_header)
    return user


@app.get("/users/me")

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")

/**
 * A stub dependency graph that is custom-made for testing concurrent build graph evaluations.
 * <p>
 * Implements a graph as follows:
 * A has no dependencies
 * B depends on A
 * C depends on A
 * X depends on B &amp; C
 * Y depends on B
 * Z depends on C
 * </p>
 *
 */
public class ProjectDependencyGraphStub implements ProjectDependencyGraph {
    public static final MavenProject A = new MavenProject();

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons=Depends(CommonQueryParams)):
    response = {}
    if commons.q:

from fastapi import Depends, FastAPI, Security
from fastapi.security import APIKeyCookie
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

api_key = APIKeyCookie(name="key", description="An API Cookie Key")


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(api_key)):
    user = User(username=oauth_header)
    return user

from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: str = Depends(get_username, scope="function")):

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")