panda/jdy_fastapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
073f0646a1767c289f82f2210a0856c5ccd378ed
jdy_fastapi/FASTAPI_QUICK_REFERENCE.md
T
panda 073f0646a1 简道云fastapi
2025-11-07 17:48:49 +08:00

9.0 KiB
Raw Blame History

FastAPI 快速参考指南

🚀 快速开始

最小示例

from fastapi import FastAPI

app = FastAPI()

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

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

运行应用

# 方式 1：使用 uvicorn
uvicorn main:app --reload

# 方式 2：在代码中运行
if __name__ == '__main__':
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

📝 路由定义

HTTP 方法

@app.get("/items")      # GET 请求
@app.post("/items")     # POST 请求
@app.put("/items/{id}") # PUT 请求
@app.delete("/items/{id}") # DELETE 请求
@app.patch("/items/{id}")  # PATCH 请求

路径参数

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 多个路径参数
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: int):
    return {"user_id": user_id, "item_id": item_id}

查询参数

@app.get("/items")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 可选参数
@app.get("/items")
async def read_items(q: str = None):
    return {"q": q}

请求体

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    description: str = None  # 可选字段

@app.post("/items")
async def create_item(item: Item):
    return item

🔧 请求和响应

获取请求数据

from fastapi import Request

@app.post("/webhook")
async def webhook(request: Request):
    # JSON 数据
    data = await request.json()
    
    # 请求头
    headers = request.headers
    
    # 查询参数
    query_params = request.query_params
    
    # 表单数据
    form_data = await request.form()
    
    # 文件上传
    files = await request.form()

响应类型

from fastapi.responses import JSONResponse, HTMLResponse, PlainTextResponse

@app.get("/json")
async def json_response():
    return JSONResponse({"message": "Hello"})

@app.get("/html")
async def html_response():
    return HTMLResponse("<h1>Hello</h1>")

@app.get("/text")
async def text_response():
    return PlainTextResponse("Hello")

状态码

from fastapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

✅ 数据验证（Pydantic）

基本模型

from pydantic import BaseModel, Field

class User(BaseModel):
    name: str
    age: int = Field(gt=0, le=120)  # 年龄必须大于 0，小于等于 120
    email: str = Field(..., regex="^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$")

嵌套模型

class Address(BaseModel):
    street: str
    city: str

class User(BaseModel):
    name: str
    address: Address

列表和字典

class Item(BaseModel):
    tags: list[str] = []
    metadata: dict[str, str] = {}

🔐 依赖注入

基本依赖

from fastapi import Depends

def get_db():
    db = "database"
    yield db
    # 清理代码

@app.get("/items")
async def read_items(db: str = Depends(get_db)):
    return {"db": db}

类依赖

class Database:
    def get_data(self):
        return "data"

def get_database():
    return Database()

@app.get("/items")
async def read_items(db: Database = Depends(get_database)):
    return db.get_data()

🛡️ 错误处理

HTTP 异常

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

自定义异常处理器

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"message": str(exc)}
    )

🔄 异步编程

异步函数

import asyncio

@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(1)  # 模拟异步操作
    return {"message": "Done"}

在线程池中运行同步函数

import anyio

def sync_function(data):
    # 同步耗时操作
    return result

@app.post("/sync-in-async")
async def sync_in_async(data: dict):
    result = await anyio.to_thread.run_sync(sync_function, data)
    return result

📊 应用状态和生命周期

应用状态

# 设置状态
app.state.my_data = "value"

# 获取状态
my_data = app.state.my_data

启动和关闭事件

@app.on_event("startup")
async def startup_event():
    # 启动时执行
    print("应用启动")

@app.on_event("shutdown")
async def shutdown_event():
    # 关闭时执行
    print("应用关闭")

🎯 中间件

添加中间件

@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

CORS 中间件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

📚 项目中的实际应用

1. 应用初始化（main.py）

app = FastAPI(title="简道云FastAPI服务")

@app.on_event("startup")
def on_startup():
    app.state.app_tools = AppTools(Config)
    app.state.logger = setup_global_logger(Config)
    app.state.f6_module = F6Module()

2. 路由处理

@app.post("/webhook")
async def webhook(request: Request):
    # 获取状态
    logger = app.state.logger
    app_tools = app.state.app_tools
    
    # 获取请求数据
    data = await request.json()
    header = request.headers
    
    # 处理逻辑
    # ...
    
    # 返回响应
    return JSONResponse(result)

3. 任务队列集成

# 将任务放入队列
response_queue = app_tools.enqueue_task(handler, data)

# 等待结果（在线程池中执行）
result = await anyio.to_thread.run_sync(response_queue.get)

🔍 调试技巧

1. 查看自动生成的文档

启动应用后访问：

2. 打印请求信息

@app.post("/webhook")
async def webhook(request: Request):
    print(f"Method: {request.method}")
    print(f"URL: {request.url}")
    print(f"Headers: {dict(request.headers)}")
    data = await request.json()
    print(f"Body: {data}")

3. 使用日志

import logging

logger = logging.getLogger(__name__)

@app.post("/webhook")
async def webhook(request: Request):
    logger.info("收到请求")
    logger.debug(f"请求数据: {data}")

📖 常用模式

1. 操作映射模式（项目中使用）

def get_action_map() -> dict:
    return {
        'action1': handler1,
        'action2': handler2,
    }

@app.post("/api")
async def api(request: Request):
    action_map = get_action_map()
    action = request.headers.get('Action')
    handler = action_map.get(action)
    if handler:
        result = handler(data)
        return JSONResponse(result)

2. 统一错误处理

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(f"未处理的异常: {exc}")
    return JSONResponse(
        status_code=500,
        content={"message": "内部服务器错误"}
    )

3. 请求验证

from pydantic import BaseModel, validator

class WebhookData(BaseModel):
    api_key: str
    entry_id: str
    data_id: str
    
    @validator('api_key')
    def validate_api_key(cls, v):
        if not v:
            raise ValueError('api_key 不能为空')
        return v

@app.post("/webhook")
async def webhook(data: WebhookData):
    # 数据已自动验证
    return {"received": data.dict()}

🎓 学习检查清单

  • 理解 FastAPI 基本概念
  • 能够创建简单的路由
  • 理解异步编程（async/await）
  • 掌握 Pydantic 数据验证
  • 理解依赖注入
  • 能够处理错误
  • 理解应用状态和生命周期
  • 能够添加中间件
  • 理解项目中的任务队列机制
  • 能够添加新的路由和功能

📚 相关资源

提示: 这个快速参考指南可以作为日常开发的速查手册。建议结合 FASTAPI_LEARNING.md 深入学习。

Reference in New Issue View Git Blame Copy Permalink