为了深入了解FastAPI框架，笔者决定将其基础库Starlette的源码进行阅读，未来也可能阅读uvicorn库的ASGI原理。个人水平十分有限，有错误的地方还请见谅。

Starlette文件结构

从applications.py开始

applications只包含了Starlette一个类

 class Starlette:
  """
        创建一个APP
        参数:
            (1) debug: 布尔值, 用于设置在出现错误时是否应返回调试回溯.

            (2) routers: 一个路由列表, 提供HTTP和WebSocket服务.

            (3) middleware: 一个中间件列表, 应用于每个request请求.
                一个starlette应用, 最少包含两个中间件:
                1. ServerErrorMiddleware中间件处于最外层, 以处理在整个堆栈中任何地方发生的未捕获错误.
                2. ExceptionMiddleware在最内层以处理处理路由或终结点中发生的异常情况.

            (4) exception_handlers: 一个字典, 键为状态码或者异常类, 值为其对应的调用函数.
                调用函数应该是handler(request, exc) -> response, 可以是标准函数, 也可以是异步函数.

            (5) on_startup: 一个启动时的调用项列表, 启动处理调用项, 不接受任何参数, 可以是同步函数, 也可以是异步函数.

            (6) on_shutdown: 一个关闭时的调用项列表, 同上.
    """
    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[["Starlette"], typing.AsyncGenerator] = None,
    ) -> None:
        """
            lifespan上下文函数是一种较新的样式，
            它取代了on_startup/on_shutdown处理程序。
            两种方式任选其一，切勿一起使用。
        """
        assert lifespan is None or (
            on_startup is None and on_shutdown is None
        ), "Use either 'lifespan' or 'on_startup'/'on_shutdown', 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()
        # build_middleware_stack函数用于构建中间件堆栈

    def build_middleware_stack(self) -> ASGIApp:
        """
            构建中间件堆栈
            返回值是ServerErrorMiddleware, 也就是最外层的中间件.其app属性指向上一个中间件, 以此类推
        """
        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,
                )
            ]
        )
        # 创建中间件元祖, 顺序为:
        # (1)最外层ServerErrorMiddleware(异常类)
        # (2)用户的中间件列表
        # (3)最内层ExceptionMiddleware(状态码字典)

        app = self.router
        for cls, options in reversed(middleware):
            app = cls(app=app, **options)
        # cls, options为middleware类的两个属性
        # 分别对应上面Middleware的第一个参数, 和后续参数.
        # 随后进行套娃, 迭代中第一个app中的app属性为self.router, 随后每个中间件app中的app属性, 皆指向上一个中间件app
        # 把最后一个套娃(应该是最外层的ServerErrorMiddleware)返回

        return app

    @property
    def routes(self) -> typing.List[BaseRoute]:
        return self.router.routes

    @property
    def debug(self) -> bool:
        return self._debug

    @debug.setter
    def debug(self, value: bool) -> None:
        self._debug = value
        self.middleware_stack = self.build_middleware_stack()

    def url_path_for(self, name: str, **path_params: str) -> URLPath:
        return self.router.url_path_for(name, **path_params)

App的启动入口

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        scope["app"] = self
        # 将自身装入到传过来的scope字典中, 然后传给中间件堆栈
        await self.middleware_stack(scope, receive, send)
        # 前面提到, middleware_stack实际上是最外层中间件ServerErrorMiddleware
        # 调用ServerErrorMiddleware的__call__

errors.py → ServerErrorMiddleware类

与上文相关的部分

class ServerErrorMiddleware:
    """
        当服务器发生错误时, 返回500响应

        如果debug设置了, 那么将返回traceback, 否则将调用指定的handler

        这个中间件类通常应该用于包含其他所有中间件, 这样保证任何地方出现异常
        都可以返回 500 response
    """

    def __init__(
        self, app: ASGIApp, handler: typing.Callable = None, debug: bool = False
    ) -> None:
        self.app = app
        self.handler = handler
        self.debug = debug

        # if key in (500, Exception):
        #     error_handler = value
        # [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug,)]
        # 中间件堆栈定义时的参数, handler为用户自定义的500处理:

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        # 如果请求类型不是http, 则转入下一个中间件

        response_started = False
        async def _send(message: Message) -> None:
            nonlocal response_started, send

            if message["type"] == "http.response.start":
                response_started = True
            await send(message)
        # 做一个闭包传给下一个中间件, 让后续中间件使用send时,
        # 能够修改自身的response_started为True

        try:
            await self.app(scope, receive, _send)
        except Exception as exc:
            # 如果调用过_send, response_started为True
            # 下面代码将不会执行
            if not response_started:
                request = Request(scope)
                if self.debug:
                    # 在debug模式中, 将返回traceback
                    response = self.debug_response(request, exc)
                elif self.handler is None:
                    # 使用默认的handler处理
                    response = self.error_response(request, exc)
                else:
                    # 使用用户自定义的500 handler
                    if asyncio.iscoroutinefunction(self.handler):
                        response = await self.handler(request, exc)
                    else:
                        response = await run_in_threadpool(self.handler, request, exc)
                    # 判断是否为协程, 不是则以线程池方式运行
                await response(scope, receive, send)

            # 我们总是不断地抛出异常。
            # 这允许服务器记录错误，
            # 或者允许测试客户端在测试用例中选择性地提出错误。
            raise exc from None

调用

uvicorn.run(app, host="127.0.0.1", port=8000)
运行时将app传给ASGI服务器, 服务器调用时会沿着
app.__call__ → middleware1.__call__ → middleware2.__call__...
查看最内层中间件ExceptionMiddleware.__call__的部分代码

    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:

会发现和最外层的ServerErrorMiddleware有极为相似, 暂且推测. 整个中间件堆栈中, 多次使用了将send闭包传给下层, 下层可能再做一次闭包将自己的response_started也包括进去.

其结构抽象化如下:

    async def send3(message):
        nonlocal response_started
        if message["type"] == "http.response.start":
            response_started = True
            # ExceptionMiddleware.response_started
        await def send2(message):
                nonlocal response_started
                if message["type"] == "http.response.start":
                    response_started = True
                    # XXXMiddleware.response_started
                await def send1(message):
                        nonlocal response_started
                        if message["type"] == "http.response.start":
                            response_started = True
                            # ServerErrorMiddleware.response_started
                        await send(message)

回到applications.py → Starlette类

英文注释在这里做了警告, 以下都是修改Starlette自身属性的方法

    # The following usages are now discouraged in favour of configuration
    #  during Starlette.__init__(...)
    # 以下方法, 现在不支持在__init__中配置使用
    def on_event(self, event_type: str) -> typing.Callable:
        return self.router.on_event(event_type)

    def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
        self.router.mount(path, app=app, name=name)

    def host(self, host: str, app: ASGIApp, name: str = None) -> None:
        self.router.host(host, app=app, name=name)

    def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
        self.user_middleware.insert(0, Middleware(middleware_class, **options))
        self.middleware_stack = self.build_middleware_stack()

    def add_exception_handler(
        self,
        exc_class_or_status_code: typing.Union[int, typing.Type[Exception]],
        handler: typing.Callable,
    ) -> None:
        self.exception_handlers[exc_class_or_status_code] = handler
        self.middleware_stack = self.build_middleware_stack()

    def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
        self.router.add_event_handler(event_type, func)

    def add_route(
        self,
        path: str,
        route: typing.Callable,
        methods: typing.List[str] = None,
        name: str = None,
        include_in_schema: bool = True,
    ) -> None:
        self.router.add_route(
            path, route, methods=methods, name=name, include_in_schema=include_in_schema
        )

    def add_websocket_route(
        self, path: str, route: typing.Callable, name: str = None
    ) -> None:
        self.router.add_websocket_route(path, route, name=name)

以及对上述方法的一些包装

    def exception_handler(
        self, exc_class_or_status_code: typing.Union[int, typing.Type[Exception]]
    ) -> typing.Callable:
        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_exception_handler(exc_class_or_status_code, func)
            return func

        return decorator

    def route(
        self,
        path: str,
        methods: typing.List[str] = None,
        name: str = None,
        include_in_schema: bool = True,
    ) -> typing.Callable:
        def decorator(func: typing.Callable) -> typing.Callable:
            self.router.add_route(
                path,
                func,
                methods=methods,
                name=name,
                include_in_schema=include_in_schema,
            )
            return func

        return decorator

    def websocket_route(self, path: str, name: str = None) -> typing.Callable:
        def decorator(func: typing.Callable) -> typing.Callable:
            self.router.add_websocket_route(path, func, name=name)
            return func

        return decorator

    def middleware(self, middleware_type: str) -> typing.Callable:
        assert (
            middleware_type == "http"
        ), 'Currently only middleware("http") is supported.'

        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_middleware(BaseHTTPMiddleware, dispatch=func)
            return func

        return decorator

以上为Starlette库中applications.py为主的源码解读, 如有错误欢迎指正.