🔧 Implementation Deep Dive
The FastASGI Application Class
The FastASGI class in
fastasgi/fastasgi.py is the single entry-point that the
ASGI server calls. Its __call__ method receives the three
ASGI primitives — scope, receive, and
send — and dispatches to the appropriate handler based on
the protocol type:
async def __call__(self, scope, receive, send): if scope["type"] == "http": await self._handle_http(scope, receive, send) elif scope["type"] == "lifespan": await self._handle_lifespan(scope, receive, send) else: await self._handle_unsupported_protocol(send)
This is the ASGI callable interface in action. The server (e.g.
Uvicorn) calls await app(scope, receive, send) for every
incoming connection, and the framework decides what to do with it.
HTTP Request Lifecycle
The _handle_http method orchestrates the entire
request-response cycle in five steps:
-
Build the Request object —
Request.from_asgi(scope, receive)wraps the raw ASGI scope and reads the complete body from thereceivechannel. - Run the middleware stack — The pre-built middleware chain is invoked. It processes the request through each middleware before reaching the router.
- Route and handle — The router matches the path and method, injects parameters, and calls the handler function.
-
Send the ASGI response — The
Responseobject is converted to ASGI messages (http.response.startfollowed byhttp.response.body) and sent back via thesendcallable. -
Clean up — Temporary files from uploads are deleted
in a
finallyblock so resources are released even when an exception occurs.
Lifespan Protocol Handling
FastASGI registers its own internal startup handler that builds the middleware chain, and a shutdown handler that cleans up any active request resources. User-registered handlers are appended after these internal ones, so the framework is always ready before user code runs.
weakref.WeakSet:
The Request class keeps a global WeakSet of
all active request objects. Because weak references do not prevent
garbage collection, requests that go out of scope are automatically
removed from the set. During shutdown the framework iterates this set
to clean up any lingering temp files.
Middleware Chain — Design & Algorithm
The middleware system is implemented in
fastasgi/middleware/middlewarechain.py. It follows the
classic "onion" model: each middleware wraps the next
one, so a request travels inward through all layers, reaches the
router at the centre, and the response travels back outward through
the same layers in reverse order.
Why Function-Composition Chaining?
There are two common approaches to building a middleware pipeline:
- List iteration — store middleware in a list and loop through them at request time. Simple, but every request pays the cost of iterating the list and deciding "what's next."
- Function composition — at startup, wrap each middleware around the next one to produce a single callable. Subsequent requests simply call that callable with zero lookup overhead.
FastASGI uses function composition. The
MiddlewareChain.build() method runs once during the
lifespan startup event and produces one async function that embodies
the entire pipeline. After that, every HTTP request calls that
function directly — no list traversal, no index tracking.
The Chaining Algorithm
The algorithm starts from the innermost layer (the router's
handle_request) and wraps outward:
def build(self, endpoint): if not self._middlewares: return endpoint current_handler = endpoint # Walk middleware in reverse so the first-registered # middleware becomes the outermost wrapper. for middleware in reversed(self._middlewares): next_handler = current_handler async def middleware_handler( request, mw=middleware, next_app=next_handler ): async def call_next(req): return await next_app(req) return await mw(request, call_next) current_handler = middleware_handler return current_handler
If you register middleware [A, B, C], the build loop processes them in reversed order C → B → A and produces a chain equivalent to:
Request → A → B → C → router → C → B → A → Response
mw=middleware and
next_app=next_handler default arguments are critical.
Without them, every closure would capture the same loop
variable and the chain would break. This is a well-known Python
pitfall when creating closures inside loops — default arguments bind
the value at definition time, not at call time.
Protocol class:
The MiddlewareCallable type is defined using Python's
typing.Protocol, which enables structural subtyping (duck
typing with type-checker support). Any async function matching the
signature (request, call_next) → Response satisfies the
protocol without explicitly inheriting from it.
Routing Engine — Design & Algorithm
Routing in FastASGI is handled by two classes:
Route (individual route definition) in
fastasgi/routing/route.py and
APIRouter (route collection and dispatch) in
fastasgi/routing/apirouter.py.
Route Matching Strategy
FastASGI uses a linear scan with regex matching. When a request comes in, the router iterates through its sorted list of routes and returns the first match. Each individual route compiles its path pattern into a regular expression at registration time, so the per-request cost is just a regex match — not a pattern parse.
Path Pattern Compilation
When you define a route like
/users/{user_id:int}/posts/{post_id}, the
Route.__init__ method calls
_compile_route_pattern() which walks the path string
character by character:
-
Literal segments like
/users/are regex-escaped and appended verbatim. -
Parameter segments inside
{...}are parsed for a name and an optional type (int,float,uuid,multipath, orstrby default). The appropriate regex capture group is emitted — for example(\d+)forintor([^/]+)forstr.
The result is a single compiled re.Pattern object and an
ordered dictionary mapping parameter names to their Python types.
Because the regex is compiled once and reused on every request, the
amortised cost is very low.
Segment-Count Optimisation
Before running the regex, the matches() method performs a
quick segment count check: it compares the number of
/-separated segments in the request path with the route's
precomputed count. If they differ (and the route does not contain a
multipath parameter) the regex is skipped entirely. This fast-path
rejection avoids expensive regex calls for the majority of
non-matching routes.
Specificity-Based Sorting
Every time a route is added, the list is re-sorted by a specificity tuple:
# (priority, literal_segments, total_segments, -definition_order) # Higher values → checked first
- priority — explicit user-defined value (highest precedence).
- literal_segments — more static segments means more specific.
- total_segments — longer paths are more specific than shorter ones.
- definition_order — among equally specific routes, the one registered first wins (negated so earlier = higher).
This ensures that /users/profile is always checked before
/users/{id} without requiring the developer to assign
manual priorities in most cases.
Automatic Parameter Injection
FastASGI inspects handler function signatures at route-registration
time using Python's inspect module. This up-front work
allows the framework to:
-
Detect which parameters are path parameters and which expect a
Requestobject, based on type annotations. - Validate at startup that every path parameter in the route pattern has a corresponding handler parameter, and vice versa — catching mismatches early instead of at request time.
-
Build a fast injection path: at request time,
route.handle()simply populates akwargsdict from the precomputed lists and calls the handler.
# At registration time: sig = inspect.signature(handler) # → identifies request_params, handler_path_params, expected_path_params # At request time: kwargs = {} for name in self.request_params: kwargs[name] = request for name in self.expected_path_params: kwargs[name] = request.path_params[name] return await self.handler(**kwargs)
inspect.signature:
The inspect module lets you introspect a function's
parameter names, default values, and type annotations at runtime.
FastASGI uses this to build a mapping between URL path parameters and
handler function arguments, enabling the clean
async def get_user(user_id: int, request: Request) style
without any boilerplate.
Request Object — Lazy Parsing & Properties
The Request class in
fastasgi/request/request.py wraps the raw ASGI scope and
body. It uses two key design patterns:
Async Factory Method
Python's __init__ cannot be async, but loading the
request body requires awaiting the receive callable.
FastASGI solves this with the
async factory method pattern:
# Can't do this — __init__ can't be async: # request = Request(scope, receive) # body not loaded yet # Instead, use the factory method: request = await Request.from_asgi(scope, receive) # → calls __init__ then awaits load_body()
The from_asgi class method creates the instance and then
calls await request.load_body(), which loops over the
receive callable until more_body is
False. This gives the rest of the framework a fully
loaded, synchronous-access request object.
Lazy-Parsed Properties
Parsing headers, query strings, cookies, and JSON is deferred until
first access using Python's @property decorator with a
private cache field:
@property def headers(self) -> Dict[str, str]: if self._headers is None: self._headers = {} for name, value in self._scope.get("headers", []): self._headers[name.decode().lower()] = value.decode() return self._headers
This means a handler that only needs request.path never
pays the cost of parsing cookies or decoding JSON. The first call
populates the cache; subsequent calls return the cached value
instantly.
Multipart Parsing — Approach & Trade-offs
File uploads arrive as multipart/form-data content. The
parser lives in fastasgi/request/multipart/parser.py.
FastASGI's implementation takes a deliberate
simplicity-first
approach:
The Non-Streaming Approach
FastASGI collects the entire request body into memory first, then parses it as a complete byte string. The algorithm is:
-
Extract the boundary string from the
Content-Typeheader. -
Split the body bytes on
--{boundary}to isolate individual parts. -
For each part, split on
\r\n\r\nto separate headers from content. -
Parse the
Content-Dispositionheader to determine field name and (optionally) filename. -
If a filename is present, write the content to a temporary file and
wrap it in an
UploadFileobject. Otherwise, decode it as a form field.
# Core parsing logic (simplified) boundary_bytes = f"--{boundary}".encode() parts = body.split(boundary_bytes) for part in parts[1:-1]: # skip first empty + last closing headers, content = part.split(b"\r\n\r\n", 1) # → parse Content-Disposition, extract field/file
FastASGI reads the complete message body before parsing. This
simplifies the implementation enormously because you can use
bytes.split() and simple string operations. However, it
means the entire upload must fit in memory at once.
A production-grade approach would parse the body
as it arrives from the ASGI
receive channel — processing each chunk and writing
file data to disk incrementally. This streaming strategy can handle
uploads of any size with constant memory, but it is
significantly more complex to implement: you must
maintain parser state across chunks, handle boundaries that span two
chunks, and manage partial header reads. Frameworks like Starlette
use the
python-multipart
library for this purpose.
The UploadFile Class
Parsed files are wrapped in UploadFile objects that
provide a clean interface:
-
open(mode="rb")— returns a standard Python file handle for reading. Write modes are explicitly rejected to protect the temp file. -
save(path)— copies the temp file to a permanent location usingshutil.copy2(preserving metadata). -
cleanup()— deletes the temp file. Called automatically by theRequestin itsfinallyblock and also in__del__as a safety net.
Response Building — Content Detection & ASGI Conversion
The Response class in
fastasgi/response.py accepts content in several Python
types and automatically determines the right
Content-Type header:
-
dictorlist→ serialised withjson.dumps, content type set toapplication/json. -
str→ scanned for HTML markers like<html>or<h1>. If found, the type istext/html; otherwisetext/plain. -
bytes→ kept as-is, content type defaults toapplication/octet-stream. -
int/float→ converted to string, typetext/plain.
An explicit content_type argument always overrides auto-
detection. The convenience functions text_response(),
json_response(), html_response(), and
redirect_response() are thin wrappers that set the
content type explicitly and make handler code more readable.
Converting to ASGI Messages
to_asgi_response() converts the high-level
Response into the dictionary format the
FastASGI._send_response method expects. It encodes every
header name and value to bytes (as required by the ASGI spec) and
appends each Set-Cookie value as a separate header entry
— because HTTP allows multiple Set-Cookie headers but not
comma-separated values.
Method Chaining
set_header(), set_cookie(), and
delete_cookie() all return self, enabling a
fluent interface:
return Response("OK")\ .set_header("X-Custom", "value")\ .set_cookie("session", "abc123", httponly=True)
This pattern is common in builder-style APIs and keeps response construction concise.
Python Techniques Summary
| Technique | Where Used | Why |
|---|---|---|
| Closure with default args | MiddlewareChain.build() | Capture loop variables at definition time |
typing.Protocol |
MiddlewareCallable | Structural subtyping without inheritance |
| Async factory method | Request.from_asgi() | __init__ cannot be async |
Lazy @property |
Request.headers, .cookies, .query_params | Avoid parsing cost until needed |
weakref.WeakSet |
Request._active_requests | Track instances without preventing GC |
inspect.signature |
Route._inspect_handler_signature() | Introspect handler params for injection |
Compiled re.Pattern |
Route._compile_route_pattern() | One-time compile, fast per-request match |
| Tuple sorting key | APIRouter._calculate_route_specificity() | Multi-criteria route ordering |
Method chaining (return self) |
Response.set_header(), .set_cookie() | Fluent builder-style API |