Unlocking FastAPI: A Deep Dive into Starlette, ASGI, and Middleware Architecture

FastAPI is essentially an API framework that wraps Starlette; to fully understand FastAPI you must first understand Starlette.

1. ASGI Protocol

Uvicorn interacts with an ASGI application through a generic interface. An application can send and receive messages with Uvicorn by implementing the following code:

<code>async def app(scope, receive, send):
    assert scope['type'] == 'http'
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ]
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!'
    })

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")
</code>

2. Starlette

To start Starlette with Uvicorn you can use the following code:

<code>from starlette.applications import Starlette
from starlette.middleware.gzip import GZipMiddleware

app = Starlette()

@app.route("/")
def demo_route():
    pass

@app.websocket_route("/")
def demo_websocket_route():
    pass

@app.add_exception_handlers(404)
def not_found_route():
    pass

@app.on_event("startup")
def startup_event_demo():
    pass

@app.on_event("shutdown")
def shutdown_event_demo():
    pass

app.add_middleware(GZipMiddleware)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000)
</code>

This code initializes Starlette, registers routes, exception handlers, events, and middleware, then passes the application to uvicorn.run . The uvicorn.run method calls Starlette's call method to send request data.

Analyzing Starlette's initialization:

<code>class Starlette:
    def __init__(self,
                 debug: bool = False,
                 routes: typing.Sequence[BaseRoute] = None,
                 middleware: typing.Sequence[Middleware] = None,
                 exception_handlers: typing.Dict[typing.Union[int, typing.Type[Exception]], typing.Callable] = None,
                 on_startup: typing.Sequence[typing.Callable] = None,
                 on_shutdown: typing.Sequence[typing.Callable] = None,
                 lifespan: typing.Callable[[self], typing.AsyncGenerator] = None) -> None:
        """
        :param debug: Enable debug mode.
        :param routes: List of routes providing HTTP and WebSocket services.
        :param middleware: List of middleware applied to each request.
        :param exception_handlers: Mapping of HTTP status codes to callbacks.
        :param on_startup: Callbacks executed on startup.
        :param on_shutdown: Callbacks executed on shutdown.
        :param lifespan: ASGI lifespan function.
        """
        assert lifespan is None or (on_startup is None and on_shutdown is None), "Only 'lifespan' or 'on_startup'/'on_shutdown' can be used, not both."
        self._debug = debug
        self.state = State()
        self.router = Router(routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan)
        self.exception_handlers = {} if exception_handlers is None else dict(exception_handlers)
        self.user_middleware = [] if middleware is None else list(middleware)
        self.middleware_stack = self.build_middleware_stack()
</code>

The middleware‑building function is defined as:

<code>class Starlette:
    def build_middleware_stack(self) -> ASGIApp:
        debug = self.debug
        error_handler = None
        exception_handlers = {}
        for key, value in self.exception_handlers.items():
            if key in (500, Exception):
                error_handler = value
            else:
                exception_handlers[key] = value
        middleware = (
            [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
            + self.user_middleware
            + [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)]
        )
        app = self.router
        for cls, options in reversed(middleware):
            app = cls(app=app, **options)
        return app
</code>

After building the middleware stack, uvicorn.run calls the call method:

<code>class Starlette:
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        scope["app"] = self
        await self.middleware_stack(scope, receive, send)
</code>

This method sets the application in the request scope and then invokes the middleware stack, demonstrating that every component in Starlette is designed as an ASGI app.

2. Middleware

In Starlette, middleware is an ASGI app, so every middleware class must follow this pattern:

<code>class BaseMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        pass

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        pass
</code>

The starlette.middleware package contains many implementations, but this article focuses on a few representative ones.

2.1 ExceptionMiddleware

ExceptionMiddleware is not used directly by developers; instead, developers register callbacks with @app.exception_handlers(status_code) . When an exception occurs, ExceptionMiddleware looks up the appropriate handler and returns its result.

<code>class ExceptionMiddleware:
    def __init__(self, app: ASGIApp, handlers: dict = None, debug: bool = False):
        self.app = app
        self.debug = debug
        self._status_handlers = {}
        self._exception_handlers = {HTTPException: self.http_exception}
        if handlers is not None:
            for key, value in handlers.items():
                self.add_exception_handler(key, value)

    def add_exception_handler(self, exc_class_or_status_code, handler) -> None:
        if isinstance(exc_class_or_status_code, int):
            self._status_handlers[exc_class_or_status_code] = handler
        else:
            assert issubclass(exc_class_or_status_code, Exception)
            self._exception_handlers[exc_class_or_status_code] = handler

    def _lookup_exception_handler(self, exc: Exception) -> typing.Optional[typing.Callable]:
        for cls in type(exc).__mro__:
            if cls in self._exception_handlers:
                return self._exception_handlers[cls]
        return None

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        response_started = False
        async def sender(message: Message) -> None:
            nonlocal response_started
            if message["type"] == "http.response.start":
                response_started = True
            await send(message)
        try:
            await self.app(scope, receive, sender)
        except Exception as exc:
            handler = None
            if isinstance(exc, HTTPException):
                handler = self._status_handlers.get(exc.status_code)
            if handler is None:
                handler = self._lookup_exception_handler(exc)
            if handler is None:
                raise exc
            if response_started:
                raise RuntimeError("Caught handled exception, but response already started.")
            request = Request(scope, receive=receive)
            if asyncio.iscoroutinefunction(handler):
                response = await handler(request, exc)
            else:
                response = await run_in_threadpool(handler, request, exc)
            await response(scope, receive, sender)
</code>

2.2 User Middleware

When developers create custom middleware they usually subclass BaseHTTPMiddleware and implement the dispatch method. Code before the call to call_next runs before the request is processed, and code after runs after the response is generated.

<code>class DemoMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp):
        super(DemoMiddleware, self).__init__(app)

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        response: Response = await call_next(request)
        return response
</code>

The underlying implementation of BaseHTTPMiddleware looks like this:

<code>class BaseHTTPMiddleware:
    def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None):
        self.app = app
        self.dispatch_func = self.dispatch if dispatch is None else dispatch

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """ASGI entry point for HTTP requests."""
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        request = Request(scope, receive=receive)
        response = await self.dispatch_func(request, self.call_next)
        await response(scope, receive, send)

    async def call_next(self, request: Request) -> Response:
        loop = asyncio.get_event_loop()
        queue: "asyncio.Queue[typing.Optional[Message]]" = asyncio.Queue()
        scope = request.scope
        receive = request.receive
        send = queue.put
        async def coro():
            try:
                await self.app(scope, receive, send)
            finally:
                await queue.put(None)
        task = loop.create_task(coro())
        message = await queue.get()
        if message is None:
            task.result()
            raise RuntimeError("No response returned.")
        assert message["type"] == "http.response.start"
        async def body_stream() -> typing.AsyncGenerator[bytes, None]:
            while True:
                message = await queue.get()
                if message is None:
                    break
                assert message["type"] == "http.response.body"
                yield message.get("body", b"")
                task.result()
</code>

2.3 ServerErrorMiddleware

ServerErrorMiddleware is similar to ExceptionMiddleware but acts as a fallback to guarantee a valid HTTP response. Its logic is:

If debug mode is enabled, return the debug page.

If a callback is registered for the error, execute it.

Otherwise, return a generic 500 response.

3. Routing

Starlette separates routing into two parts: the core router (under the middleware layer) that handles most of the framework logic, and the individual routes registered with the router.

3.1 Router

The router is straightforward; its main responsibilities are loading and matching routes.

<code>class Router:
    def __init__(self,
                 routes: typing.Sequence[BaseRoute] = None,
                 redirect_slashes: bool = True,
                 default: ASGIApp = None,
                 on_startup: typing.Sequence[typing.Callable] = None,
                 on_shutdown: typing.Sequence[typing.Callable] = None,
                 lifespan: typing.Callable[[typing.Any], typing.AsyncGenerator] = None) -> None:
        self.routes = [] if routes is None else list(routes)
        self.redirect_slashes = redirect_slashes
        self.default = self.not_found if default is None else default
        self.on_startup = [] if on_startup is None else list(on_startup)
        self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)

    async def default_lifespan(self, app: typing.Any) -> typing.AsyncGenerator:
        await self.startup()
        yield
        await self.shutdown()

    async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Logic executed when no route matches."""
        if scope["type"] == "websocket":
            websocket_close = WebSocketClose()
            await websocket_close(scope, receive, send)
            return
        if "app" in scope:
            raise HTTPException(status_code=404)
        else:
            response = PlainTextResponse("Not Found", status_code=404)
            await response(scope, receive, send)

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Entry point for the Router class."""
        assert scope["type"] in ("http", "websocket", "lifespan")
        if "router" not in scope:
            scope["router"] = self
        if scope["type"] == "lifespan":
            await self.lifespan(scope, receive, send)
            return
        partial = None
        for route in self.routes:
            match, child_scope = route.matches(scope)
            if match == Match.FULL:
                scope.update(child_scope)
                await route.handle(scope, receive, send)
                return
            elif match == Match.PARTIAL and partial is None:
                partial = route
                partial_scope = child_scope
        if partial is not None:
            scope.update(partial_scope)
            await partial.handle(scope, receive, send)
            return
        if scope["type"] == "http" and self.redirect_slashes and scope["path"] != "/":
            redirect_scope = dict(scope)
            if scope["path"].endswith("/"):
                redirect_scope["path"] = redirect_scope["path"].rstrip("/")
            else:
                redirect_scope["path"] = redirect_scope["path"] + "/"
            for route in self.routes:
                match, child_scope = route.matches(redirect_scope)
                if match != Match.NONE:
                    redirect_url = URL(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    await response(scope, receive, send)
                    return
        await self.default(scope, receive, send)
</code>

The router's __call__ method performs the matching loop, handling full matches first, then partial matches, and finally optional slash redirects. Performance tests show that for fewer than 50 routes the simple loop is faster than a routing tree, and up to 100 routes the difference is negligible.

3.2 Other Routes

Starlette provides several route classes that inherit from BaseRoute :

<code>class BaseRoute:
    def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
        raise NotImplementedError()

    def url_path_for(self, name: str, **path_params: str) -> URLPath:
        raise NotImplementedError()

    async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
        raise NotImplementedError()

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Routes can be used as standalone ASGI applications."""
        match, child_scope = self.matches(scope)
        if match == Match.NONE:
            if scope["type"] == "http":
                response = PlainTextResponse("Not Found", status_code=404)
                await response(scope, receive, send)
            elif scope["type"] == "websocket":
                websocket_close = WebSocketClose()
                await websocket_close(scope, receive, send)
            return
        scope.update(child_scope)
        await self.handle(scope, receive, send)
</code>

The concrete subclasses are:

Route : Standard HTTP route matched by URL and method.

WebSocketRoute : Handles WebSocket connections using starlette.websocket.WebSocket .

Mount : Prefix‑based nesting that forwards requests to another ASGI app, enabling route grouping and composition such as Router → Mount → Router → Route .

Host : Dispatches requests based on the Host header to different ASGI apps.

4. Other Components

Most Starlette components are designed as ASGI applications, which maximizes compatibility at a modest performance cost.

<code>├── middleware
├── applications.py
├── authentication.py
├── background.py
├── concurrency.py
├── config.py
├── convertors.py
├── datastructures.py
├── endpoints.py
├── exceptions.py
├── formparsers.py
├── graphql.py
├── __init__.py
├── py.typed
├── requests.py
├── responses.py
├── routing.py
├── schemas.py
├── staticfiles.py
├── status.py
├── templating.py
├── testclient.py
├── types.py
└── websockets.py
</code>

Some of the simpler files are omitted from this overview.

5. Summary

We have examined several core parts of Starlette, including its ASGI foundation, middleware system, routing mechanisms, and overall project structure. Starlette’s design emphasizes extensibility and compatibility, making it an excellent reference for building your own web frameworks.