Httpapi Module API Reference¶

def allow_only(*allowed: str) -> Callable[[Request], Awaitable[None]]:
    """Create a dependency that rejects unknown query parameters (HTTP 422).

    Args:
      *allowed (str): The set of query parameter names that are permitted.

    Returns:
      Callable[[Request], Awaitable[None]]: An async dependency suitable for
      FastAPI's ``dependencies=[...]``. It raises an ``HTTPException`` with
      status 422 if the request includes parameters outside the allowlist.

    Raises:
      HTTPException: Emitted by the returned dependency at request time when
        unknown query parameters are present (422 Unprocessable Entity).
    """
    allowed_set: set[str] = set(allowed)

    async def _dep(request: Request) -> None:
        """Raise an HTTPException if unknown query parameters are present.

        This function is designed to be used as a FastAPI dependency. It validates
        that the request's query string contains only parameters from a pre-defined
        set of allowed names, assumed to be in the parent scope's `allowed_set`
        variable.

        Args:
            request: The incoming FastAPI/Starlette request object.

        Raises:
            HTTPException: An exception with a 422 status code and a
                problem+json body if any query parameters are found that are
                not in the `allowed_set`.
        """
        extras = set(request.query_params.keys()) - allowed_set
        if extras:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=Problem(
                    type=AnyUrl("https://bijux-cli.dev/docs/errors/validation-error"),
                    title="Validation error",
                    status=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail=f"Unknown query params: {', '.join(sorted(extras))}",
                    instance=str(request.url),
                ).model_dump(mode="json"),
            )

    return _dep

@router.post(
    "/items",
    response_model=Item,
    status_code=status.HTTP_201_CREATED,
    summary="Create item",
    description="Create a new item.",
    responses={
        200: {
            "model": Item,
            "description": "Item already exists; existing resource returned",
        },
        406: {"model": Problem},
        409: {"model": Problem},
        422: {"model": Problem},
    },
    tags=["Items"],
    dependencies=[Depends(require_accept_json)],
)
def create_item(
    response: Response,
    item: ItemIn = Body(...),  # noqa: B008
    store: ItemStoreProtocol = Depends(get_store),  # noqa: B008
) -> Item:
    """Creates a new item.

    Args:
        item (ItemIn): The data for the new item from the request body.
        store (ItemStoreProtocol): The dependency-injected item store.

    Returns:
        Item: The newly created item, including its server-generated ID.
    """
    existing = store.find_by_name(item.name)
    if existing is not None:
        response.status_code = status.HTTP_200_OK
        return existing
    return store.create(item)

@router.delete(
    "/items/{item_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Delete item",
    description="Delete an item by its ID.",
    responses={
        406: {"model": Problem},
        404: {"model": Problem},
        422: {"model": Problem},
    },
    tags=["Items"],
    dependencies=[Depends(require_accept_json)],
)
def delete_item(
    item: Annotated[Item, Depends(get_item_or_404)],
    store: Annotated[ItemStoreProtocol, Depends(get_store)],
) -> Response:
    """Delete an item by its unique ID.

    The target item is resolved by the `get_item_or_404` dependency before this
    handler runs.

    Args:
        item: The item to delete, injected by `get_item_or_404`.
        store: The item store implementation, injected.

    Returns:
        Response: Empty body with **204 No Content** on successful deletion.

    Raises:
        HTTPException: 404 if the item does not exist (raised by the dependency).
    """
    store.delete(item.id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

@router.get(
    "/items/{item_id}",
    response_model=Item,
    summary="Get item",
    description="Get a single item by its ID.",
    responses={404: {"model": Problem}, 406: {"model": Problem}},
    tags=["Items"],
    dependencies=[Depends(require_accept_json)],
)
def get_item(
    item: Item = Depends(get_item_or_404),  # noqa: B008
) -> Item:
    """Retrieves a single item by its ID.

    This endpoint uses a dependency (`get_item_or_404`) to fetch the item,
    ensuring that a 404 response is returned if the item does not exist.

    Args:
        item (Item): The item retrieved by the `get_item_or_404` dependency.

    Returns:
        Item: The requested item.
    """
    return item

def get_item_or_404(
    item_id: int = Path(..., ge=1),
    store: ItemStoreProtocol = Depends(get_store),  # noqa: B008
) -> Item:
    """A dependency that retrieves an item by ID or raises a 404."""
    return store.get(item_id)

def get_store() -> ItemStoreProtocol:
    """A FastAPI dependency to provide the `ItemStoreProtocol` instance."""
    return store

@app.get("/health", summary="Health check", tags=["Health"])
async def health() -> dict[str, str]:
    """Lightweight readiness probe used by Makefile `api-test`."""
    return {"status": "ok"}

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
    """A custom exception handler for `HTTPException`.

    This handler intercepts FastAPI's standard HTTP exceptions and ensures they
    are logged and returned in the standard JSON error format.

    Args:
        request (Request): The incoming request.
        exc (HTTPException): The HTTP exception.

    Returns:
        JSONResponse: A JSON response detailing the HTTP error.
    """
    logger.warning("HTTP error: %s %s", exc.status_code, exc.detail)
    return JSONResponse(status_code=exc.status_code, content=exc.detail)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Manages the application's lifespan events for startup and shutdown.

    On startup, this context manager resets and prepopulates the in-memory
    store with demo data. On shutdown, it resets the store again.

    Args:
        app (FastAPI): The FastAPI application instance.

    Yields:
        None: Yields control to the application while it is running.
    """
    store.reset()
    store.prepopulate(
        [
            {"name": "Item One", "description": "Description one"},
            {"name": "Item Two", "description": "Description two"},
        ]
    )
    logger.info("Prepopulated store with demo items")
    yield
    store.reset()
    logger.info("Store reset on shutdown")

@router.get(
    "/items",
    response_model=ItemListResponse,
    summary="List items",
    description="List all items with pagination.",
    tags=["Items"],
    responses={406: {"model": Problem}, 422: {"model": Problem}},
    dependencies=[
        Depends(require_accept_json),
        Depends(allow_only("limit", "offset")),
        reject_duplicate_query_params("limit", "offset"),
    ],
)
def list_items(
    limit: int = Query(10, ge=1, le=100),
    offset: int = Query(0, ge=0),
    store: ItemStoreProtocol = Depends(get_store),  # noqa: B008
) -> ItemListResponse:
    """Retrieves a paginated list of items.

    Args:
        limit (int): The maximum number of items per page.
        offset (int): The starting offset for the item list.
        store (ItemStoreProtocol): The dependency-injected item store.

    Returns:
        ItemListResponse: An object containing the list of items and total count.
    """
    items, total = store.list_items(limit, offset)
    return ItemListResponse(items=items, total=total)

def reject_duplicate_query_params(*params: str) -> DependsMarker:
    """Create a dependency that rejects duplicate query parameters (HTTP 422).

    Args:
      *params: Names of query parameters that must not appear more than once.

    Returns:
      fastapi.params.Depends: A dependency marker that, when executed at
      request time, raises ``HTTPException`` (422) if any listed parameter
      appears more than once.

    Raises:
      HTTPException: Emitted at request time if duplicates are detected.
    """

    async def _dep(request: Request) -> None:
        """Raise an HTTPException if specific query parameters are duplicated.

        This function is designed to be used as a FastAPI dependency. It checks an
        iterable of parameter names (assumed to be in the parent scope's `params`
        variable) to ensure they are not repeated in the request's query string.

        Args:
            request: The incoming FastAPI/Starlette request object.

        Raises:
            HTTPException: An exception with a 422 status code and a
                problem+json body if any of the specified query parameters
                are found more than once.
        """
        duplicates = [p for p in params if len(request.query_params.getlist(p)) > 1]
        if duplicates:
            detail = f"Duplicate query params found: {', '.join(sorted(duplicates))}"
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail=Problem(
                    type=AnyUrl("https://bijux-cli.dev/docs/errors/validation-error"),
                    title="Validation error",
                    status=status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail=detail,
                    instance=str(request.url),
                ).model_dump(mode="json"),
            )

    return DependsMarker(_dep)

def require_accept_json(request: Request) -> None:
    """Reject requests that don't accept application/json (HTTP 406).

    Schemathesis' negative-data checks may send unsupported Accept headers.
    If the client doesn't accept JSON (and not */*), respond with 406.
    """
    accept = request.headers.get("accept", "*/*").lower()
    if "*/*" in accept or "application/json" in accept:
        return
    raise HTTPException(
        status_code=status.HTTP_406_NOT_ACCEPTABLE,
        detail=Problem(
            type=AnyUrl("https://bijux-cli.dev/docs/errors/not-acceptable"),
            title="Not Acceptable",
            status=status.HTTP_406_NOT_ACCEPTABLE,
            detail="Set 'Accept: application/json' for this endpoint",
            instance=str(request.url),
        ).model_dump(mode="json"),
    )

@router.put(
    "/items/{item_id}",
    response_model=Item,
    summary="Update item",
    description="Update an existing item.",
    responses={
        406: {"model": Problem},
        404: {"model": Problem},
        409: {"model": Problem},
        422: {"model": Problem},
    },
    tags=["Items"],
    dependencies=[Depends(require_accept_json)],
)
def update_item(
    item: Annotated[Item, Depends(get_item_or_404)],
    update_data: Annotated[ItemIn, Body(...)],
    store: Annotated[ItemStoreProtocol, Depends(get_store)],
) -> Item:
    """Update an existing item.

    Args:
      item (Item): The current item resolved from the path parameter,
        injected by ``get_item_or_404``.
      update_data (ItemIn): The new values for the item (request body).
      store (ItemStoreProtocol): The item store implementation (injected).

    Returns:
      The updated item.

    Raises:
      HTTPException: If the item does not exist (404) or if the new name
        conflicts with another item (409). These are raised by the dependency
        or the store layer.
      RequestValidationError: If the path/body validation fails (422). Handled
        by the global validation exception handler.
    """
    return store.update(item.id, update_data)

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request, exc: RequestValidationError
) -> JSONResponse:
    """A custom exception handler for `RequestValidationError`.

    This handler intercepts validation errors from FastAPI and formats them
    into a standard `JSONResponse` with a 422 status code.

    Args:
        request (Request): The incoming request.
        exc (RequestValidationError): The validation exception.

    Returns:
        JSONResponse: A JSON response detailing the validation error.
    """
    logger.warning("Validation error: %s", exc.errors())
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=Problem(
            type=AnyUrl("https://bijux-cli.dev/docs/errors/validation-error"),
            title="Validation error",
            status=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=str(exc),
            instance=str(request.url),
        ).model_dump(mode="json"),
    )