HTTP Extractors — Aquilia Documentation
Comprehensive guide and documentation for HTTP Extractors in the Aquilia framework. View API reference, examples, and implementation patterns.
Dependency Injection / Extractors HTTP Parameter Extractors Bind incoming HTTP metadata directly to your dependency parameters with the built-in Header, Query, Cookie, Path, and Body extractors. Values are cast and validated through the Contract facet pipeline. How Extractors Work When the RequestDAG resolves dependencies, it checks if any parameter is annotated with an extractor dataclass. If it is, the DAG intercepts the resolution and reads the value straight from the request: from typing import Annotated from aquilia.di import Header, Query, Dep from aquilia.controller import Controller, get async def search_telemetry( user_agent: Annotated[str, Header("User-Agent")], search_query: Annotated[str, Query("q", default="")] ): print(f"Tracking search: from ") return search_query class SearchController(Controller): # Recommended constructor injection for app-wide services def __init__(self, telemetry_client: TelemetryClient): self.telemetry = telemetry_client @get("/search") async def search_view( self, ctx, query: Annotated[str, Dep(search_telemetry)] ): await self.telemetry.track("search_run") return Automatic coercion. Extracted raw strings are cast to the annotated type through the Contract facet pipeline — Annotated[int, Query("page")] yields a real int, not a string. A failed cast returns a structured BadRequestFault (HTTP 400). All five extractors accept alias to map a differently-named source key. Header Extracts an HTTP header. Lookups are case-insensitive. required defaults to True. from aquilia.di import Header @dataclass(frozen=True) class Header: name: str # header name, e.g. "Authorization" alias: str | None = None # alternate lookup key required: bool = True # missing -> BadRequestFault (HTTP 400) default: Any = None # fallback when not required # Usage async def auth(token: Annotated[str, Header("Authorization")]) -> str: return token.removeprefix("Bearer ") Query Extracts a query-string value (?key=value). required defaults to False. from aquilia.di import Query @dataclass(frozen=True) class Query: name: str | None = None # query key, e.g. "page" default: Any = None # value when absent required: bool = False # missing + required -> BadRequestFault alias: str | None = None # alternate key # Usage — cast to int with a default async def page(n: Annotated[int, Query("page", default=1)]) -> int: return n Cookie Extracts a cookie value. required defaults to False. from aquilia.di import Cookie @dataclass(frozen=True) class Cookie: name: str | None = None default: Any = None required: bool = False alias: str | None = None # Usage async def sess(sid: Annotated[str, Cookie("session_id")]) -> str: return sid Path Extracts a route/path parameter. required defaults to True. from aquilia.di import Path @dataclass(frozen=True) class Path: name: str | None = None default: Any = None required: bool = True alias: str | None = None # Usage — matches @get("/users/ "), cast to int async def load(user_id: Annotated[int, Path()]) -> int: return user_id Body Injects the parsed request body. Pair with a Contract type for full validation. from aquilia.di import Body @dataclass(frozen=True) class Body: media_type: str = "application/json" embed: bool = False # Usage async def create(data: Annotated[dict, Body()]) -> dict: return data Error Handling A missing required value, a null where null is disallowed, or a failed type cast raises BadRequestFault. The Fault Engine renders it as a structured HTTP 400 automatically — you never write the 400 yourself. RequestDAG Patterns & Recipes )
Go to Homepage