8 篇文章带有标签 “FastAPI”

APIKeyHeader 的源码

class APIKeyHeader(APIKeyBase):
    def __init__(
        self,
        *,
        name: str,
        scheme_name: Optional[str] = None,
        description: Optional[str] = None,
        auto_error: bool = True,
    ):
        self.model: APIKey = APIKey(
            **{"in": APIKeyIn.header},  # type: ignore[arg-type]
            name=name,
            description=description,
        )
        self.scheme_name = scheme_name or self.__class__.__name__
// ...

如果 auto_error=True（默认），那么当 api_key 不存在时，会抛出 HTTPException 异常，返回 403 状态码。

如果 auto_error=False，那么当 api_key 不存在时，会返回 None。那么自定义函数 get_api_key 才会被调用。

@app.get("/", response_class=HTMLResponse, include_in_schema=False)
async def index():
    # 需要过滤的路由
    filted_routes = [
        "/openapi.json",
        "/docs",
        "/docs/oauth2-redirect",
        "/redoc",
        "/static",
        "/"
    ]
    
    routes = []
    for route in app.routes:
        if route.path not in filted_routes:
// ...

pip install markdown

使用

from markdown import markdown
print(markdown("# Title"))

<h1>Title</h1>

from pydantic import BaseModel, Json, ValidationError
from fastapi import APIRouter, File, UploadFile, HTTPException, Form, Depends

class Box(BaseModel):
    x: int
    y: int
    w: int
    h: int

router = APIRouter()

@router.post("/test")
async def test(file: UploadFile = File(...), 
               mask: Json = Form(default=None), 
               n: int = Form(default=0)) -> str:
// ...

.vscode/launch.json

{
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python: FastAPI",
            "type": "python",
            "request": "launch",
            "module": "uvicorn",
            "args": [
                "app.main:app",
                "--reload"
            ],
            "jinja": true,
            "justMyCode": true
        }
    ]
}

• --reload 修改后可以自动加载，适用于开发。
• Debug FastAPI in VS Code IDE
• Debug FastAPI application in VSCode

docker build docker buildx build 如何构建多架构 Docker 镜像？

main.py

import uvicorn

from fastapi import FastAPI

from .routers import users
from .routers import files


app = FastAPI(title='REST API Interface', version='1.0', 
              description="基于 FastAPI 的 REST API 接口。")


app.include_router(users.router)
app.include_router(files.router)


@app.get('/')
async def index():
    return {'Hello': 'World!'}


if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=8000)

dependencies.

服务器 CPU 40核，内存 256G，操作系统 Ubuntu 20.04，Python3.9

wrk 的 lua 脚本：postfile_formdata.lua

wrk.method = "POST"
local f = io.open("postdata", "rb")
wrk.body   = f:read("*all")
wrk.headers["Content-Type"] = "multipart/form-data; boundary=gouchicao0123456789"

wrk 的 lua 脚本：postfile_json.lua

wrk.method = "POST"
local f = io.open("postdata.json", "rb")
wrk.body   = f:read("*all")
wrk.headers["Content-Type"] = "application/json"

/file_benchmarking/upload/binary/chunk/async_func/async_rw wrk -c100

使用 FastAPI 实现了文件的上传和下载，部署服务使用了 uvicorn 和 gunicorn+uvicorn 两种方法。

基准测试工具使用的是 wrk

服务器 CPU 40核，内存 256G，操作系统 Ubuntu 20.04，Python3.9

stream 异步读取上传的文件，同步写入 tempfile.NamedTemporaryFile() 生成的文件。 http://172.16.33.159:8000/files/upload/stream/async_read_and_memory_write stream 异步读取上传的文件，同步写入磁盘文件。 http://172.16.33.159:8000/files/upload/stream/async_read_and_disk_write stream 异步读取上传的文件，异步写入磁盘文件。 http://172.16.33.159:8000/files/upload/stream/async_read_and_async_write 异步全量读取上传的文件，异步写入磁盘文件。 http://172.16.33.159:8000/files/upload/single 异步全量读取上传的文件，异步写入磁盘文件。(API 函数定义时没有使用 async) http://172.16.33.

健康码识别服务使用了 FastAPI 进行开发的，本周主要工作是为了对健康码识别的服务进行性能调优。接口函数使用了 async 关键字，但是内部的实现并没有使用 await。由于改写成异步代码需要时间，这里并没有改写代码，只是删除了 async 关键字。部署服务使用了 uvicorn 和 gunicorn+uvicorn 两种方法。

基准测试工具使用的是 ab

• 4 个进程可以发挥到最佳效果
• 8 个进程已经到了上限了
• 在部署这种密集计算的应用下，gunicorn + uvicorn 并没有比 uvicorn 强，但如果您需要管理进程，它们就是最佳组合。
• 通过基准测试发现，最大的瓶颈不是 GPU，而且 CPU，GPU 一张卡的负载还没有 40 核 CPU 的负载高。

异步（使用了 async 关键字）函数，在压测的过程中基本上不会失败（Failed）,同步函数，在压测过程中会经常失败，随着并发数的增加而增加。目前还没有找到原因