FastAPI源码分析-Hello World

关键词：FastAPI starlette ASGIApp pydantic

FastAPI源码文件结构

查看__init__.py

可以看出，FastAPI包含以下几个模块：

• 依赖模块

starlette的status模块

• 自有模块

FastAPI、BackgroundTasks、UploadFile、HTTPException、Request、Response、APIRouter、WebSocket和param_functions内的多个模块

模块和类的命名语义化，阅读起来很顺畅，可以大致推断出其各个模块用途，以自有模块为例：

主要包括套接字（WebSocket，端口绑定与监听）、HTTP模块（HTTPException、Request、Response，http协议支持）、URL路由寻径（APIRouter，处理路由转发）、参数处理（param_functions，进行参数校验）、任务管理（BackgroundTasks）、文件IO（UploadFile）

代码

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

分析

 from fastapi import FastAPI

导入FastAPI模块，用于初始化，查看FastAPI类的源码接口（使用SourceTail分析）

app = FastAPI()

# applications.py
class FastAPI(Starlette):
    def __init__(
        self,
        *,
        debug: bool = False,
        routes: List[BaseRoute] = None,
        title: str = "FastAPI",
        description: str = "",
        version: str = "0.1.0",
        openapi_url: Optional[str] = "/openapi.json",
        openapi_prefix: str = "",
        default_response_class: Type[Response] = JSONResponse,
        docs_url: Optional[str] = "/docs",
        redoc_url: Optional[str] = "/redoc",
        swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
        swagger_ui_init_oauth: Optional[dict] = None,
        middleware: Sequence[Middleware] = None,
        exception_handlers: Dict[Union[int, Type[Exception]], Callable] = None,
        on_startup: Sequence[Callable] = None,
        on_shutdown: Sequence[Callable] = None,
        **extra: Dict[str, Any],
    ) -> None:
        self.default_response_class = default_response_class
        self._debug = debug
        self.state = State()
        self.router: routing.APIRouter = routing.APIRouter(
            routes,
            dependency_overrides_provider=self,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
        )
        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()

        self.title = title
        self.description = description
        self.version = version
        self.openapi_url = openapi_url
        self.openapi_prefix = openapi_prefix.rstrip("/")
        self.docs_url = docs_url
        self.redoc_url = redoc_url
        self.swagger_ui_oauth2_redirect_url = swagger_ui_oauth2_redirect_url
        self.swagger_ui_init_oauth = swagger_ui_init_oauth
        self.extra = extra
        self.dependency_overrides: Dict[Callable, Callable] = {}

        self.openapi_version = "3.0.2"

        if self.openapi_url:
            assert self.title, "A title must be provided for OpenAPI, e.g.: 'My API'"
            assert self.version, "A version must be provided for OpenAPI, e.g.: '2.1.0'"

        if self.docs_url or self.redoc_url:
            assert self.openapi_url, "The openapi_url is required for the docs"
        self.openapi_schema: Optional[Dict[str, Any]] = None
        self.setup()

可以看到：

1. app继承自starlette类

2. 关键参数-title、description、version、openapi_url、docs_url、redoc_url，可以用于配置自定义接口文档

3. 初始化路由router

这里有几个关键参数：

routes,on_startup,on_shutdown

其中on_startup和on_shutdown会在服务器启动和关闭时调用

@app.get("/")
def read_root():
    return {"Hello": "World"}

当使用GET方法访问指定路径时，会调用get方法，查看源码如下：

# applications.py
    def get(
        self,
        path: str,
        *,
        response_model: Type[Any] = None,
        status_code: int = 200,
        tags: List[str] = None,
        dependencies: Sequence[Depends] = None,
        summary: str = None,
        description: str = None,
        response_description: str = "Successful Response",
        responses: Dict[Union[int, str], Dict[str, Any]] = None,
        deprecated: bool = None,
        operation_id: str = None,
        response_model_include: Union[SetIntStr, DictIntStrAny] = None,
        response_model_exclude: Union[SetIntStr, DictIntStrAny] = set(),
        response_model_by_alias: bool = True,
        response_model_skip_defaults: bool = None,
        response_model_exclude_unset: bool = False,
        include_in_schema: bool = True,
        response_class: Type[Response] = None,
        name: str = None,
        callbacks: List[routing.APIRoute] = None,
    ) -> Callable:
        if response_model_skip_defaults is not None:
            warning_response_model_skip_defaults_deprecated()  # pragma: nocover
        return self.router.get(
            path,
            response_model=response_model,
            status_code=status_code,
            tags=tags or [],
            dependencies=dependencies,
            summary=summary,
            description=description,
            response_description=response_description,
            responses=responses or {},
            deprecated=deprecated,
            operation_id=operation_id,
            response_model_include=response_model_include,
            response_model_exclude=response_model_exclude,
            response_model_by_alias=response_model_by_alias,
            response_model_exclude_unset=bool(
                response_model_exclude_unset or response_model_skip_defaults
            ),
            include_in_schema=include_in_schema,
            response_class=response_class or self.default_response_class,
            name=name,
            callbacks=callbacks,
        )

• 若配置response_model_skip_defaults参数，则会调用utils类的warning_response_model_skip_defaults_deprecated方法

# utils.py
# TODO: remove when removing support for Pydantic < 1.0.0
def warning_response_model_skip_defaults_deprecated() -> None:
    logger.warning(  # pragma: nocover
        "response_model_skip_defaults has been deprecated in favor of "
        "response_model_exclude_unset to keep in line with Pydantic v1, support for "
        "it will be removed soon."
    )

也就是说，Pydantic版本低于1.0.0时，该参数不可用

• 调用GET方法，接着会router.get方法，再调用router.api_route方法，再调用router.add_api_route方法，使用传入的方法即装饰器下面的方法read_root()

总结路径

1. 启动服务器

初始化路由路径、API文档，调用on_startup

2. 访问指定路由路径

调用为：FastAPI.get->routing.APIRouter.get->routing.APIRouter.api_route->根据装饰器注入相应方法->routing.APIRouter.add_api_route->starlette.router相应方法

3. 关闭服务器

调用on_shutdown，清理资源

简而言之，当使用FastAPI的GET方法时，真正使用的是starlette的GET方法

进入starlette源码后，会看到调用Route.handle方法，再调用ASGIApp.request_response方法，然后调用concurrency.run_in_threadpool方法

若是使用同步，则直接执行传入的方法，若是使用异步，则调用标准库asyncio在线程池中运行传入的方法

总结，@app装饰器的目的是将相应路由路径和方法传给asgiapp，在asgi服务器启动后根据路由调用方法。

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

对于传入的path和query、body等参数的解析，则会调用get_request_handler方法，然后调用dependencies.utils.solve_dependencies方法，再调用request_params_to_args方法，根据参数是否必选、类型是否正确等进行校验，如果使用了BaseModel，则还会使用pydantic进行参数解析

简而言之，传参的解析和校验，若是直接传参，则使用自有utils类方法进行解析校验，若是使用参数模型，则会用到pydantic进行解析校验。

参考资料

1. FastAPI官网 ：https://fastapi.tiangolo.com/
2. FastAPI源码 ：https://github.com/tiangolo/fastapi

原文链接:https://blog.csdn.net/weixin_42078760/article/details/104872557