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}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):

    return user


async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

                } catch (InterruptedException e) {
                    log.debug("Async notification monitoring interrupted for {}", registrationId);
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.debug("Error in async notification monitoring for {}: {}", registrationId, e.getMessage());

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}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: 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: str = Header()):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

inner_strict = APIRouter(prefix="/strict", strict_content_type=True)
inner_default = APIRouter(prefix="/default")


@inner_strict.post("/items/")
async def inner_strict_post(data: dict):
    return data


@inner_default.post("/items/")
async def inner_default_post(data: dict):
    return data


outer_router.include_router(inner_strict)
outer_router.include_router(inner_default)

from starlette.websockets import WebSocket

app = FastAPI()
app.state.value = 42


async def extract_value_from_http_connection(conn: HTTPConnection):
    return conn.app.state.value


@app.get("/http")
async def get_value_by_http(value: int = Depends(extract_value_from_http_connection)):
    return value


@app.websocket("/ws")
async def get_value_by_ws(

    responses={404: {"description": "Not found"}},
)


fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}


@router.get("/")
async def read_items():
    return fake_items_db


@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")

            return response

        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=TimedRoute)


@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}

from inline_snapshot import snapshot

app = FastAPI()


@app.get("/default")
async def default(foo: Annotated[str, Query()] = "foo"):
    return {"foo": foo}


@app.get("/required")
async def required(foo: Annotated[str, Query(min_length=1)]):
    return {"foo": foo}


@app.get("/multiple")
async def multiple(foo: Annotated[str, object(), Query(min_length=1)]):
    return {"foo": foo}