FastAPI 源码阅读 (一) ASGI应用
本章开启FastAPI的源码阅读，FastAPI是当下python web中一颗新星，是一个划时代的框架。从诞生便是以快速和简洁为核心理念。
它继承于Starlette，是在其基础上的完善与扩展。详细内容可以翻看我之前的源码阅读。
目录结构
阅读方案
概要
我们可以将模块分为三类
- FastAPI的核心原创内容，这是我们的重点
- 在Starlette的基础上增加少量内容，如果未使用到，我们将放在后面
- 完全继承于Starlette的内容，这部分不再赘述
从applications.py开始
FastAPI 类
方法
openapi()与setup()是在初始化阶段，对OpenAPI文档进行初始化的函数。
而add_api_route()一直到trace()，是关于路由的函数，它们都是直接对router的方法传参引用。所以这些放在解读routing.py时一并进行。
class FastAPI(Starlette):
def __init__(
self,
*,
debug: bool = False,
routes: Optional[List[BaseRoute]] = None,
title: str = "FastAPI",
description: str = "",
version: str = "0.1.0",
openapi_url: Optional[str] = "/openapi.json",
openapi_tags: Optional[List[Dict[str, Any]]] = None,
servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
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: Optional[Sequence[Middleware]] = None,
exception_handlers: Optional[
Dict[Union[int, Type[Exception]], Callable]
] = None,
on_startup: Optional[Sequence[Callable]] = None,
on_shutdown: Optional[Sequence[Callable]] = None,
openapi_prefix: str = "",
root_path: str = "",
root_path_in_servers: bool = True,
**extra: Any,
) -> None:
"""
# starlette原生
:param debug: debug模式
:param middleware: 中间件列表
:param exception_handlers: 异常对应处理的字典
:param on_startup: 启动项列表
:param on_shutdown: 结束项列表
:param routes: 路由列表
# OpenAPI文档相关
:param docs_url: API文档地址
:param title: 标题
:param description: 描述
:param version: API版本
:param openapi_url: openapi.json的地址
:param openapi_tags: 上述内容的元数据模式
# 文档的页面中的OAuth，有关JS，以后介绍
:param swagger_ui_oauth2_redirect_url:
:param swagger_ui_init_oauth:
# Redoc文档
:param redoc_url: 文档地址
# 反向代理情况下的文档
:param servers: 服务器列表
:param openapi_prefix: 支持反向代理和挂载子应用程序，已被弃用
:param root_path: 如果有反向代理，让app直到自己"在哪"
:param root_path_in_servers: 允许自动包含root_path
:param default_response_class: 默认的response类
:param extra:
"""
self.default_response_class = default_response_class
self._debug = debug
self.state = State()
# 这里路由用的是APIRouter，和starlette所采用的不同
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.servers = servers or []
self.openapi_url = openapi_url
self.openapi_tags = openapi_tags
# TODO: remove when discarding the openapi_prefix parameter
if openapi_prefix:
logger.warning(
'openapi_prefix“已被弃用，取而代之的是更接近ASGI标准的“root_path”，它更简单，也更自动化。'
"请阅读文档: "
"https://fastapi.tiangolo.com/advanced/sub-applications/"
)
self.root_path = root_path or openapi_prefix
self.root_path_in_servers = root_path_in_servers
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'"
self.openapi_schema: Optional[Dict[str, Any]] = None
self.setup()
除了Starlette原生的参数，大量参数都是和API文档相关。
而路由从Starlette的Router换成了新式的APIRouter
关于root_path和servers
这两个概念查询了大量文档才搞明白。他们主要是关于文档与反向代理的参数。当使用了Nginx时等反向代理时，从Uvicorn直接访问，和从Nginx代理访问，路径可能出现不一致。比如Nginx中的Fastapi根目录是127.0.0.1/api/，而Uvicorn角度看是127.0.0.1:8000/。对于API接口来说，其实这个是没有影响的，因为服务器会自动帮我们解决这个问题。但对于API文档来说，就会出现问题。
因为当我们打开/docs时，网页会寻找openapi.json。他的是写在html内部的，而不是变化的。这会导致什么问题？
未经过反向代理
例如当我们从Uvicorn访问127.0.0.1:8000/docs时，他会寻找/openapi.json即去访问127.0.0.1:8000/openapi.json(了解前端的应该知道)
经过反向代理
但是假如我们这时，从Nginx外来访问文档，假设我们这样设置Nginx:
location /api/ {
proxy_pass http://127.0.0.1:8000/;
}
我们需要访问127.0.0.1/api/docs，才能从代理外部访问。而打开docs时，我们会寻找openapi.json。
注意，openapi.json是FastAPI初始化时预置的API接口，他一定要在FastAPI的内部的存在。
所以这时，它应该在127.0.0.1/api/openapi这个位置存在。
但我们的浏览器不知道这些，他会按照/openapi.json，会去寻127.0.0.1/openapi.json这个位置。所以他不可能找到openapi.json，自然会启动失败。
这其实是openapi文档前端的问题。
root_path，是用来解决这个问题的。既然/openapi.json找不到，那我自己改成/api/openapi.json不就成了么。
root_path即是这个/api，这个是在定义时手动设置的参数。为了告诉FastAPI，它处在整个主机中的哪个位置。即告知 所在根目录。这样，FastAPI就有了"自知之明"，乖乖把这个前缀加上。来找到正确的openapi.json
还没完呢
加上了root_path，openapi.json的位置变成了/api/openapi.json。当你想重新用Uvicorn提供的地址从代理内访问时，他会去寻找哪？没错127.0.0.1:8000/api/openapi.json，但我们从代理内部访问，并不需要这个前缀，但它还是“善意”的帮我们加上了，所以这时候内部的访问失灵了。
虽然我们不大可能需要从两个位置访问这个完全一样的api文档。但这点一定要注意。
设置root_path后会存在标识
root_path就这一个用处么？
我在翻官方文档时，看到他们把root_path吹得天花乱坠，甚至弃用了openapi_prefix参数。但最后是把我弄得晕头转向。
这样要提到servers这个参数，官方首先给了这么段示例，稍作修改。
from fastapi import FastAPI, Request
app = FastAPI(
servers=[
{"url": "https://stag.example.com", "description": "Staging environment"},
{"url": "https://prod.example.com", "description": "Production environment"},
],
root_path="/api",
)
@app.get("/test")
def read_main(request: Request):
return {"message": "Hello World", "root_path": request.scope.get("root_path")}
当我们打开API文档时
多了许多内容
我们可以切换这个Servers时，底下测试接口的发送链接也会变成相应的。
/api
stag.example.com
但是记住，切换这个server，下面的接口不会发送变化，只是发送的host会改变。
这代表，虽然可以通过文档，测试多个其他主机的接口。但是这些主机和自己之间，需要拥有一致的接口。这种情况通常像在线上/开发服务器或者服务器集群中可能出现。虽然不要求完全一致，但为了这样做有实际意义，最好大体上是一致的。
但是我们看到，这是在代理外打开的，如果我们想从代理内打开，需要去掉root_path。会发生什么？
我们将root_path注释掉：
依旧可以访问
很好，我们依旧可以看到这些服务器，但是。
我们找不到自己了，我们可以在这两个服务器之间来回切换，但是无法切到自己。我们无法访问自己的接口。
如果想解决这个问题，只需要将自身手动加入到Servers中。
from fastapi import FastAPI, Request
app = FastAPI(
servers=[
{"url": "/", "description": "这是你自己哦"},
{"url": "https://stag.example.com", "description": "Staging environment"},
{"url": "https://prod.example.com", "description": "Production environment"},
],
# root_path="/api/",
)
@app.get("/test")
def read_main(request: Request):
return {"message": "Hello World", "root_path": request.scope.get("root_path")}
有了自己的一席之地
可以正常使用
以下是我做的一些笔记
root_path和servers都是关于api文档的内容，只影响文档，不影响代理内外api的访问。
root_path 可以在反向代理的情况下，让api文档确认到自己的位置
servers 可以让API文档访问多个服务器，但如果没有添加root_path就无法找到自己
root_path × servers × 非代理，不显示选项，仅访问自己。代理，找不到openapi.json
root_path v servers × 非代理，找不到openapi.json。代理，显示选项，仅访问自己
root_path × servers v 非代理，显示选项，无法访问自己。代理，找不到openapi.json
root_path v servers v 非代理，找不到openapi.json。代理，显示选项，都可以访问
- root_path 的有无 决定你能在代理内还是外访问到openapi。
- 没有servers时，默认访问自己。有servers时，按servers里的内容来。
- 如果servers没有自己，那就是无法访问自己。
- root_path非空时，会自动把自己加入到servers中，
- root_path为空时，想访问自己，请手动写'/'到servers中
- root_path_in_servers会决定 ④是否把自动把自己加入到servers中
关于root_path_in_servers，当root_path和servers都存在时，root_path会自动将自己加入到servers中。但如果这个置为False，就不会自动加入。(默认为True)
API文档初始化
class FastAPI(Starlette):
......
def openapi(self) -> Dict:
if not self.openapi_schema:
self.openapi_schema = get_openapi(
title=self.title,
version=self.version,
openapi_version=self.openapi_version,
description=self.description,
routes=self.routes,
tags=self.openapi_tags,
servers=self.servers,
)
return self.openapi_schema
def setup(self) -> None:
if self.openapi_url:
# 部署openapi.json
urls = (server_data.get("url") for server_data in self.servers)
# 例:
# servers=[
# {"url": "https://stag.example.com", "description": "Staging environment"},
# {"url": "https://prod.example.com", "description": "Production environment"},
# ],
server_urls = {url for url in urls if url}
# 把所有非空url取出
# openapi.json的endpoint
async def openapi(req: Request) -> JSONResponse:
root_path = req.scope.get("root_path", "").rstrip("/")
# root_path 为 "" 或 "/" 时， 不会被自动加入。需自行手动填写到servers中
if root_path not in server_urls:
if root_path and self.root_path_in_servers:
self.servers.insert(0, {"url": root_path})
server_urls.add(root_path)
# 如果没有且允许加入，那就把root_path加入到servers中
return JSONResponse(self.openapi())
# 添加host:port/openapi_url 这条路由，对应openapi.json。
# include_in_schema代表是否在文档中收录自己
self.add_route(self.openapi_url, openapi, include_in_schema=False)
if self.openapi_url and self.docs_url:
# 设置docs可视化文档
# docs的endpoint
async def swagger_ui_html(req: Request) -> HTMLResponse:
root_path = req.scope.get("root_path", "").rstrip("/")
openapi_url = root_path + self.openapi_url
# 拼接openapi.json的路径，这使代理内/外的一方无法访问
oauth2_redirect_url = self.swagger_ui_oauth2_redirect_url
if oauth2_redirect_url:
oauth2_redirect_url = root_path + oauth2_redirect_url
return get_swagger_ui_html(
openapi_url=openapi_url,
title=self.title + " - Swagger UI",
oauth2_redirect_url=oauth2_redirect_url,
init_oauth=self.swagger_ui_init_oauth,
)
self.add_route(self.docs_url, swagger_ui_html, include_in_schema=False)
if self.swagger_ui_oauth2_redirect_url:
# oauth认证的endpoint
async def swagger_ui_redirect(req: Request) -> HTMLResponse:
return get_swagger_ui_oauth2_redirect_html()
self.add_route(
self.swagger_ui_oauth2_redirect_url,
swagger_ui_redirect,
include_in_schema=False,
)
if self.openapi_url and self.redoc_url:
# redoc的endpoint
async def redoc_html(req: Request) -> HTMLResponse:
root_path = req.scope.get("root_path", "").rstrip("/")
openapi_url = root_path + self.openapi_url
return get_redoc_html(
openapi_url=openapi_url, title=self.title + " - ReDoc"
)
self.add_route(self.redoc_url, redoc_html, include_in_schema=False)
self.add_exception_handler(HTTPException, http_exception_handler)
self.add_exception_handler(
RequestValidationError, request_validation_exception_handler
)
API文档实际上以字符串方式，在FastAPI内部拼接的。实际上就是传统的模板(Templates)，这个相信大家都很熟悉了。优点是生成时灵活，但缺点是不容易二次开发。fastapi提供了好几种文档插件，也可以自己添加需要的。
路由的添加与装饰器
def add_api_route(
self,
path: str,
endpoint: Callable,
*,
response_model: Optional[Type[Any]] = None,
status_code: int = 200,
tags: Optional[List[str]] = None,
dependencies: Optional[Sequence[Depends]] = None,
summary: Optional[str] = None,
description: Optional[str] = None,
response_description: str = "Successful Response",
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
deprecated: Optional[bool] = None,
methods: Optional[List[str]] = None,
operation_id: Optional[str] = None,
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
response_model_by_alias: bool = True,
response_model_exclude_unset: bool = False,
response_model_exclude_defaults: bool = False,
response_model_exclude_none: bool = False,
include_in_schema: bool = True,
response_class: Optional[Type[Response]] = None,
name: Optional[str] = None,
) -> None:
self.router.add_api_route(
path,
endpoint=endpoint,
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,
methods=methods,
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=response_model_exclude_unset,
response_model_exclude_defaults=response_model_exclude_defaults,
response_model_exclude_none=response_model_exclude_none,
include_in_schema=include_in_schema,
response_class=response_class or self.default_response_class,
name=name,
)
这么长一大串，实际上就一句话self.router.add_api_route()，其他剩下的那些我暂且省略的，其实基本都是这样的。就是调用router的一个功能。下面我用省略方式将它们列出。
def add_api_route(...):
self.router.add_api_route(...)
def api_route(...):
def decorator(func: Callable) -> Callable:
self.router.add_api_route(...)
return func
return decorator
def add_api_websocket_route(
self, path: str, endpoint: Callable, name: Optional[str] = None
) -> None:
self.router.add_api_websocket_route(path, endpoint, name=name)
def websocket(self, path: str, name: Optional[str] = None) -> Callable:
def decorator(func: Callable) -> Callable:
self.add_api_websocket_route(path, func, name=name)
return func
return decorator
def include_router(...):
self.router.include_router(...)
def get(...):
return self.router.get(...)
def put(...):
return self.router.put(...)
def post(...):
return self.router.post(...)
def delete(...):
return self.router.delete(...)
def options(...):
return self.router.options(...)
def head(...):
return self.router.head(...)
def path(...):
return self.router.paht(...)
def trace(...):
return self.router.trace(...)
可以看到有些在这里就做了闭包，实际上除了这里的'add_api_route()'他们最终都是要做闭包的。只是过程放在里router里。它们最终的指向都是router.add_api_route()，这是一个添加真正将endpoint加入到路由中的方法。
FastAPI添加路由的方式，在starlette的传统路由列表方式上做了改进，变成了装饰器式。
@app.get('/path')
def endport():
return {"msg": "hello"}
其实就是通过这些方法作为装饰器，将自身作为endpoint传入生成route节点，加入到routes中。
App入口
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if self.root_path:
scope["root_path"] = self.root_path
if AsyncExitStack:
async with AsyncExitStack() as stack:
scope["fastapi_astack"] = stack
await super().__call__(scope, receive, send)
# 直接借用starlette的__call__进入中间件堆栈
else:
await super().__call__(scope, receive, send) # pragma: no cover
FastAPI的入口没有太大的变化，借用starlette的await self.middleware_stack(scope, receive, send)直接进入中间件堆栈。
