Fastapi 基础 — 木小良

文档：https://fastapi.tiangolo.com/zh/
pip install fastapi uvicorn
pip install pyjwt jwt
pip install passlib[bcrypt] 密码哈希
pip install asyncpg 如果使用异步操作pg数据库要安装
python-multipart 如果要实现上传文件需要安装

python

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
FastAPI示例应用
这个模块展示了如何创建一个基本的FastAPI应用，并生成API文档
"""

from fastapi import FastAPI, Path, Query, HTTPException, status, Body # 导入FastAPI框架的类和函数
from fastapi.middleware.cors import CORSMiddleware # 导入CORS中间件
from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html # 导入生成API文档的函数
from fastapi.openapi.utils import get_openapi # 导入生成API文档的函数
from pydantic import BaseModel, Field # 导入数据模型和字段验证的类
import uvicorn # 导入运行FastAPI应用的函数
from datetime import datetime # 导入日期时间类
from contextlib import asynccontextmanager # 导入异步上下文管理器

# 模拟数据库
items_db = {}
counter = 0

# 初始化数据
def init_data():
    # 初始化模拟数据
    global items_db, counter
    counter = 3
    items_db = {
        1: {
            "id": 1,
            "name": "iPhone 13",
            "description": "Apple最新款手机",
            "price": 5999.0,
            "tax": 599.9,
            "tags": ["手机", "苹果", "电子产品"]
        },
        2: {
            "id": 2,
            "name": "MacBook Pro",
            "description": "专业级笔记本电脑",
            "price": 12999.0,
            "tax": 1299.9,
            "tags": ["笔记本", "苹果", "电子产品"]
        },
        3: {
            "id": 3,
            "name": "iPad Pro",
            "description": "专业级平板电脑",
            "price": 6999.0,
            "tax": 699.9,
            "tags": ["平板", "苹果", "电子产品"]
        }
    }

# 生命周期管理器，用于管理应用程序的生命周期
@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    参数：
    app: FastAPI 应用程序实例
    """
    # 启动前代码
    print('启动前')
    init_data()
    # 运行期间
    yield  # 代码执行到这里时，控制权会交给FastAPI框架
    # 结束前
    print('结束前')
    items_db.clear()

# 创建FastAPI应用实例
# title: 应用标题,在api文档中显示
# description: 应用描述,在api文档中显示
# version: 应用版本,在api文档中显示
# lifespan: 生命周期管理器
app = FastAPI(
    title="示例API",
    description="这是一个FastAPI示例应用，展示API文档生成",
    version="1.0.0",
    lifespan=lifespan  # 传入生命周期管理器
)

# 配置CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 允许所有源
    allow_credentials=True, # 允许跨域请求
    allow_methods=["*"],  # 允许所有方法
    allow_headers=["*"],  # 允许所有头
)

# 定义API请求参数模型，用户验证字段
class Item(BaseModel):
    """商品模型"""
    id: int | None = None 
    name: str = Field(..., description="商品名称", min_length=1, max_length=50, examples=["示例商品", "iPhone 13"])
    description: str | None = Field(None, description="商品描述", examples=["这是一个示例商品"])
    price: float = Field(..., description="商品价格", gt=0, examples=[99.9, 5999.0])
    tax: float | None = Field(None, description="税费", examples=[10.5])
    tags: list[str] = Field(default=[], description="标签列表", examples=[["示例", "测试"]])

class ItemUpdate(BaseModel):
    """商品更新模型"""
    name: str | None = Field(None, description="商品名称", min_length=1, max_length=50)
    description: str | None = Field(None, description="商品描述")
    price: float | None = Field(None, description="商品价格", gt=0)
    tax: float | None = Field(None, description="税费")
    tags: list[str] | None = Field(None, description="标签列表")


# 路由: 首页，tags，summary，description，在api文档中显示,分别代表分组，标题，描述
@app.get("/", tags=["首页"],summary="首页内容", description="首页内容详细介绍")
async def read_root():
    # 首页直接返回一个json
    return {
        "message": "欢迎使用FastAPI示例应用",
        "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }

# 路由: 获取所有商品,response_model: 返回数据模型
@app.get("/items/", tags=["商品"], response_model=list[Item])
# skip: 初始跳过前n个商品
# limit: 限制返回的商品数量
async def read_items(skip: int = 0, limit: int = 10):
    # 返回商品列表，跳过前n个商品，并限制返回的商品数量
    return list(items_db.values())[skip:skip+limit]

# 路由: 获取单个商品
@app.get("/items/{item_id}", tags=["商品"], response_model=Item)
async def read_item(
    # 路径参数，必填，类型为int，最小值为0
    item_id: int = Path(..., description="商品ID", ge=0)
):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品未找到")
    # 返回单个商品
    return items_db[item_id]

# 路由: 创建商品
@app.post("/items/", tags=["商品"], response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    global counter
    counter += 1
    item_dict = item.dict()
    item_dict["id"] = counter
    items_db[counter] = item_dict
    return item_dict

# 路由: 更新商品
@app.put("/items/{item_id}", tags=["商品"], response_model=Item)
async def update_item(
    item_id: int = Path(..., description="商品ID", ge=0),
    item: ItemUpdate = Body(...)
):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品未找到")
        
    update_data = item.dict(exclude_unset=True)
    stored_item = items_db[item_id]
    
    for field, value in update_data.items():
        stored_item[field] = value
    
    return stored_item

# 路由: 删除商品
@app.delete("/items/{item_id}", tags=["商品"])
async def delete_item(
    item_id: int = Path(..., description="商品ID", ge=0)
):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品未找到")
    
    del items_db[item_id]
    return {"message": "商品删除成功"}

# 路由: 搜索商品
@app.get("/items/search/", tags=["商品"], response_model=list[Item])
async def search_items(
    name: str | None = Query(None, description="商品名称"),
    min_price: float | None = Query(None, description="最低价格"),
    max_price: float | None = Query(None, description="最高价格"),
    tags: list[str] | None = Query(None, description="标签列表")
):
    results = list(items_db.values())
    
    if name:
        results = [item for item in results if name.lower() in item["name"].lower()]
    
    if min_price is not None:
        results = [item for item in results if item["price"] >= min_price]
    
    if max_price is not None:
        results = [item for item in results if item["price"] <= max_price]
    
    if tags:
        results = [
            item for item in results 
            if any(tag in item.get("tags", []) for tag in tags)
        ]
    
    return results

# 自定义API文档
def custom_openapi():
    """生成自定义OpenAPI架构"""
    if app.openapi_schema:
        return app.openapi_schema
    
    openapi_schema = get_openapi(
        title="示例API",
        version="1.0.0",
        description="这是一个FastAPI示例应用，展示API文档生成",
        routes=app.routes,
    )
    
    # 自定义徽标信息
    openapi_schema["info"]["x-logo"] = {
        "url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"
    }
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

# 自定义API文档路径
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
    """自定义Swagger UI文档页面"""
    return get_swagger_ui_html(
        openapi_url=app.openapi_url,
        title=f"{app.title} - Swagger UI",
        oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
        swagger_js_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js",
        swagger_css_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css",
    )

@app.get("/redoc", include_in_schema=False)
async def redoc_html():
    """自定义ReDoc文档页面"""
    return get_redoc_html(
        openapi_url=app.openapi_url,
        title=f"{app.title} - ReDoc",
        redoc_js_url="https://cdn.jsdelivr.net/npm/redoc@next/bundles/redoc.standalone.js",
    )

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

上传文件

pythono

from fastapi import UploadFile, File

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)): 
    return {"filename": file.filename}

Fastapi+sqlalchemy长连接同步版

main.py:

python

from fastapi import FastAPI
from contextlib import asynccontextmanager
from db import create_user_in_db, get_all_users, init_db

@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield

app = FastAPI(lifespan=lifespan)

@app.post("/users/")
def create_user(name: str):
    return create_user_in_db(name)  # 直接调用 db.py 中的函数

@app.get("/users/")
def read_users():
    return get_all_users()  # 直接调用 db.py 中的函数

db.py:

python

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import os

# 数据库配置
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/testdb")

# 创建数据库引擎（带连接池，保持长连接）
engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,  # 每次获取连接时检查连接是否可用
    pool_size=10,        # 连接池最大连接数
    max_overflow=20      # 超出连接池后的最大连接数
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 基类模型
Base = declarative_base()

# 模型定义
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

# 数据库操作函数（自动管理 Session）
def create_user_in_db(name: str) -> User:
    db = SessionLocal()  # 从连接池获取连接
    try:
        user = User(name=name)
        db.add(user)
        db.commit()
        db.refresh(user)
        return user
    finally:
        db.close()  # 释放回连接池

def get_all_users():
    db = SessionLocal()  # 从连接池获取连接
    try:
        return db.query(User).all()
    finally:
        db.close()  # 释放回连接池

# 初始化数据库（创建表）
def init_db():
    Base.metadata.create_all(bind=engine)

上下文的作用：实现自动释放关闭会话

python

from contextlib import asynccontextmanager
@asynccontextmanager
async def get_db_context():
    """上下文管理器自动关闭会话"""
    db = SessionLocal()
    try:
        yield db
        await db.commit()
    except Exception:
        await db.rollback()
        raise
    finally:
        await db.close()


# 在函数中引用上下文
async def get_persona_list() -> list[PersonaModel]:
    async with get_db_context() as session:
        result = session.query(PersonaModel).filter(PersonaModel.is_active == True).all()
        return result

路由和实列分离

routes/users.py:

python

from pydantic import BaseModel, Field
from datetime import timedelta
from fastapi import  status, Response,APIRouter
from fastapi_ser.jwtser import create_access_token


# 创建路由器
router = APIRouter(
    prefix="/user",  # 路由前缀
    tags=["用户模块"]     # API文档标签
)

# 后台登录模型
class LoginModel(BaseModel):
    username: str = Field(..., description="用户名")
    password: str = Field(..., description="密码")

# 后台登录
@router.post("/login", summary="用户登录", tags=["登录"])
async def login(user_data: LoginModel, response: Response):
    """
    用户登录接口
    """
    # 验证用户名和密码
    if user_data.username != "crowdpulse" or user_data.password != "crowdpulse":
        # 设置401未授权状态码
        response.status_code = status.HTTP_401_UNAUTHORIZED
        return {'msg':'用户名或密码错误'}
    
    # 创建访问令牌
    access_token_expires = timedelta(minutes=30)
    access_token = await create_access_token(
        data={"sub": user_data.username, "username": user_data.username},
        expires_delta=access_token_expires
    )
    return {'msg':access_token}

main.py:

python

# 导入路由模块
from routes import users,hot

# 创建FastAPI实例
app = FastAPI(
    title="crowdpulse 管理后台API说明",
    description="crowdpulse 管理后台API说明",
    version="1.0.0",
    lifespan=lifespan  # 传入生命周期管理器
)

# 注册路由
app.include_router(users.router)
app.include_router(hot.router)

验证器+自定义输出错误

python

#第三方随意创建一个类
class Msg(str):
    """字段自定义错误文本的标记类，不用写内容"""
    pass

# 定义的模型：
from pydantic import BaseModel, Field, EmailStr
from typing import Annotated
from 从第三方导入一个 import Msg
class SignupModel(BaseModel):
	name:Optional[str] = Field(None, description="规则名称") #默认空值，可为空，但是字段类型为str
    username: Annotated[str,Field(min_length=5, max_length=50),Msg("username长度必须 5~50 字符")]
    email: Annotated[EmailStr,Field(description="邮箱地址"),Msg("邮箱地址格式不正确")]
    password: Annotated[str,Field(min_length=6, max_length=100),Msg("密码长度必须 6~100 字符")]

#main.py:
# 缓存函数，用于获取请求体模型
@lru_cache(maxsize=None)
def get_request_body_model(endpoint: callable) -> type[BaseModel] | None:
    """
    通过反射检查端点签名，找到用于请求体的Pydantic模型。
    结果会被缓存，以避免重复的性能开销。
    """
    try:
        # 获取端点函数的签名
        sig = inspect.signature(endpoint)
        for param in sig.parameters.values():
            # 检查参数的类型注解是否为 BaseModel 的子类
            if inspect.isclass(param.annotation) and issubclass(param.annotation, BaseModel):
                # 这是一个启发式方法：返回找到的第一个Pydantic模型。
                # 对于简单的 POST/PUT 端点通常是正确的。
                return param.annotation
    except (ValueError, TypeError):
        # 如果端点没有有效的签名（例如，来自其他WSGI/ASGI应用的转发）
        return None
    return None

	
# 全局异常处理：把第一个字段错误替换成自定义 Msg，并输出纯文本
@app.exception_handler(RequestValidationError)
async def custom_validation_handler(request: Request, exc: RequestValidationError):
    first_err = exc.errors()[0]
    field_name = first_err["loc"][-1]  # 触发错误的字段名

    # 通过反射拿到当前路由函数里的 BaseModel（此处是 Register）
    endpoint = request.scope["endpoint"]
	# 使用缓存函数获取请求体模型
    model_cls = get_request_body_model(endpoint)
    # model_cls = next(
    #     (
    #         ann.annotation
    #         for ann in signature(endpoint).parameters.values()
    #         if isclass(ann.annotation) and issubclass(ann.annotation, BaseModel)
    #     ),
    #     None,
    # )

    # 如果找到模型，则尝试提取 Msg；否则用默认 msg
    if model_cls:
        metas = model_cls.model_fields[field_name].metadata
        custom_msg = next((m for m in metas if isinstance(m, Msg)), None)
        if custom_msg:
            return PlainTextResponse(str(custom_msg), status_code=422)

    return PlainTextResponse(first_err["msg"], status_code=422)

路由依赖注入

请求到达 → 路由匹配成功
├─▶ 执行该路由声明的所有依赖（Depends）
│ ├─ 若依赖函数返回值正常 → 注入到路由参数
│ └─ 若依赖函数抛出了异常/返回 Response → 立即结束，直接把该异常或响应返给客户端
└─▶ 所有依赖都通过后，再执行真正的路由处理函数，路由处理函数可以读取依赖返回的结果

生产环境部署

pip install gunicorn
启动命令
gunicorn holdem.preset_games:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
如果要信任反向代理转发的ip地址，在启动命令后面空格+--proxy-headers --forwarded-allow-ips='*'

网关模式API(一个路由处理多种方法)

如果使用了统一的自定义错误处理，可能需要额外修改这个处理函数

from fastapi import Response,APIRouter
from fastapi_ser.routes.utils import Msg
from pydantic import BaseModel,Field,field_validator
from typing import Union,Annotated,Literal

# 创建路由器
router = APIRouter(
    tags=["回调 / 内网通信"]     # API文档标签
)

class TestModel(BaseModel):
    """
    action = "test" 的请求体模型
    需求：
      - name 必须有值，且不能为空字符串 => 自定义错误 "name不能为空"
      - number 必须 > 0，默认 1 => 自定义错误 "number必须大于0"
    """
    action: Literal["test"]  # discriminator，用于区分不同动作

    name: Annotated[
        str,
        Field(
            ...,
            min_length=1,  # 至少 1 个字符
        ),
        Msg("name不能为空"),
    ]

    number: Annotated[
        int,
        Field(
            1,  # 默认值 1
            gt=0,  # 必须 > 0
        ),
        Msg("number必须大于0"),
    ]

    @field_validator("name")
    @classmethod
    def name_not_blank(cls, v: str) -> str:
        """
        额外防一下全空格的情况（"   " 这种），也视为为空
        """
        if not v or not v.strip():
            # 这里抛出的具体错误信息最后会被我们的异常处理器替换成 Msg 里的内容
            raise ValueError("name不能为空")
        return v

# 测试2模型
class Test2Model(BaseModel):
    """
    action = "test2" 的请求体模型
    需求：
      - name 可以为空（可选字段）
      - age 必须介于 18~60 之间 => 自定义错误 "age必须介于18-60之间"
    """
    action: Literal["test2"]
    name: str | None = None  # 可选字段，允许不传或为 null
    age: Annotated[
        int,
        Field(
            ...,
            ge=18,  # >= 18
            le=60,  # <= 60
        ),
        Msg("age必须介于18-60之间"),
    ]

# 定义一个联合类型
ActionModel = Annotated[
    Union[TestModel, Test2Model],
    Field(discriminator="action"), # 使用action来区分不同动作
]

# ============== 处理具体业务函数 ==============
async def test(data: TestModel) -> str:
    """
    处理x浏览器养号
    """
    return data

async def test2(data: Test2Model) -> str:
    """
    处理x浏览器养号
    """
    return data

# 封装一个统一固定的路由接口，使用方法来区分不同动作
@router.post("/webhook", summary="回调统一网关")
async def webhook_handler(req: ActionModel,response:Response) -> dict:
    """
    回调统一网关
    """
    try:
        # 根据类型执行不同的方法
        if isinstance(req, TestModel):
            result = await test(req)
        elif isinstance(req, Test2Model):
            result = await test2(req)

        return {"code": 200, "data": result}
    except Exception as e:
        response.status_code = 500
        return {"code": 500, "msg": str(e)}