FastAPI 基础知识
1 启动
# fastapi cli
fastapi dev main.py
fastapi run main.py
# uvicorn
uvicoin main:app --reload --host 0.0.0.0 --port 8000
uvicoin main:app --host 0.0.0.0 --port 80
uvicoin main:app --host 0.0.0.0 --port 80 --workers 4
# gunicorn
gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:80
2 挂载子应用
如果需要多个独立的 FastAPI 应用，拥有各自独立的 OpenAPI 与文档，则需设置一个主应用，并挂载一个（或多个）子应用，挂载是指在特定路径中添加完全独立的应用(类似Django中的应用路由分发)。
from fastapi import FastAPI
app = FastAPI()
@app.get("/app")
def read_main():
return {"message": "Hello World from main app"}
subapi = FastAPI()
@subapi.get("/sub")
def read_sub():
return {"message": "Hello World from sub API"}
app.mount("/subapi", subapi)
3 事件处理器
FastAPI 只执行主应用中的事件处理器，不执行子应用-挂载中的事件处理器。
3.1 startup
使用 startup 事件声明 app 启动前运行的函数。只有所有 startup 事件处理器运行完毕，FastAPI 应用才开始接收请求。
from fastapi import FastAPI
app = FastAPI()
items = {}
@app.on_event("startup")
async def startup_event():
print('do something before main app start')
@app.get("/items/{item_id}")
async def read_items(item_id: str):
return items[item_id]
3.2 shutdown
使用 shutdown 事件声明 app 关闭时运行的函数。
from fastapi import FastAPI
app = FastAPI()
@app.on_event("shutdown")
def shutdown_event():
print('do something when shutdown main app')
with open("log.txt", mode="a") as log:
log.write("Application shutdown")
@app.get("/items/")
async def read_items():
return [{"name": "Foo"}]
4 中间件
4.1 自定义中间件
先定义中间件函数，再使用装饰器 @app.middleware("http")
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
4.2 官方中间件
传入请求必须是 https 或 wss，任何传向 http 或 ws 的请求都会被重定向
from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app = FastAPI()
app.add_middleware(HTTPSRedirectMiddleware)
@app.get("/")
async def main():
return {"message": "Hello World"}
传入请求都必须正确设置 Host 请求头，以防 HTTP 主机头攻击。
from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app = FastAPI()
app.add_middleware(
TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"]
)
@app.get("/")
async def main():
return {"message": "Hello World"}
处理 Accept-Encoding 请求头中包含 gzip 请求的 GZip 响应。
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)
@app.get("/")
async def main():
return "somebigcontent"
5 部署管理工具
- Docker
- Kubernetes
- Docker Compose
- Docker in Swarm mode
- Systemd
- Supervisor