Python Web API开发

Python Web API 开发就是用 Python 把业务能力包装成 HTTP 接口，让浏览器、移动端、其他后端服务、批处理任务或 AI 应用都能通过网络调用。零基础不要只记 @app.get() 这种写法，要理解“一次请求从客户端到代码再到响应”的完整链路。

本页以 FastAPI 为主，因为它类型提示清晰、自动生成接口文档、参数校验能力强，适合从零学习，也适合商业项目落地。

学习目标

学完本页你应该能回答：

1. Web API 是什么，为什么后端系统都需要 API。
2. HTTP、ASGI、Uvicorn、FastAPI 分别负责什么。
3. 一次请求在 FastAPI 中经历哪些步骤。
4. 同步接口和异步接口有什么区别，写错会有什么后果。
5. 参数校验、依赖注入、中间件、异常处理分别解决什么问题。
6. 商业项目中如何分层、鉴权、记录日志、排查慢接口。
7. 如何写一个可以运行、可以测试、可以部署的 API Demo。

为什么要学 Web API

脚本只能在本机或任务机上执行，而 Web API 可以把能力开放给其他系统。例如：

场景	API 做什么
后台管理系统	提供用户、角色、订单、审批数据接口
医疗数据采集平台	提供采集任务创建、数据查询、资产登记接口
AI 知识库	提供问答、检索、文件上传、流式输出接口
数据处理平台	提供批量导入、清洗触发、处理状态查询接口
微服务系统	让服务之间通过 HTTP 调用能力

如果不会 API 分层，最常见的问题是把参数校验、鉴权、业务判断、数据库访问、模型调用、异常处理全写在一个函数里。短期能跑，长期会出现几个问题：

1. 接口越来越长，没人敢改。
2. 单元测试很难写，只能靠人工点页面。
3. 数据库慢、AI 慢、外部接口慢时，不知道慢在哪里。
4. 异常直接暴露给前端，既不友好也不安全。
5. 业务复用困难，另一个接口想复用只能复制代码。

HTTP 请求基础

一个 HTTP 请求通常包含：

部分	示例	作用
Method	GET、POST、PUT、DELETE	表示动作
Path	/api/users/1	表示资源位置
Query	?page=1&size=20	URL 上的查询参数
Header	Authorization: Bearer xxx	鉴权、内容类型、追踪信息
Body	{"name":"Tom"}	请求体，常用于新增或修改

常见 HTTP 方法：

方法	常见含义	是否通常有 Body
GET	查询资源	一般没有
POST	新增、提交动作、复杂查询	可以有
PUT	整体更新	可以有
PATCH	局部更新	可以有
DELETE	删除	一般没有，也可按约定有

HTTP 状态码不要乱用：

状态码	含义	典型场景
200	成功	查询成功、普通处理成功
201	已创建	新增资源成功
400	请求参数错误	业务参数非法
401	未登录	Token 缺失或无效
403	无权限	已登录但没有权限
404	资源不存在	用户、订单、任务不存在
422	参数校验失败	FastAPI/Pydantic 校验不通过
500	服务内部错误	未处理异常、代码缺陷

ASGI、Uvicorn、FastAPI 的关系

很多初学者会把它们混在一起。可以这样理解：

名称	角色	类比
ASGI	Python Web 服务和应用之间的异步接口规范	插座标准
Uvicorn	ASGI Server，接收网络请求并调用应用	插座和电线
FastAPI	Web 框架，负责路由、参数校验、依赖注入、响应	业务机器
Pydantic	数据模型和参数校验	质检员

请求链路：

flowchart TD
    A["客户端发起 HTTP 请求"] --> B["Uvicorn 监听端口"]
    B --> C["把请求转换为 ASGI scope"]
    C --> D["FastAPI 路由匹配"]
    D --> E["Pydantic 参数校验"]
    E --> F["执行依赖注入"]
    F --> G["调用接口函数"]
    G --> H["返回 dict 或 Response"]
    H --> I["FastAPI 序列化为 JSON"]
    I --> J["Uvicorn 写回 HTTP 响应"]

如果没有 Uvicorn，FastAPI 代码只是一个应用对象，没有网络监听能力。如果没有 FastAPI，你需要自己解析路由、参数、异常和响应，工程成本会很高。

最小可运行 Demo

安装：

pip install fastapi uvicorn

main.py：

from fastapi import FastAPI

app = FastAPI(title="demo-api")


@app.get("/health")
def health():
    return {"status": "ok"}

启动：

uvicorn main:app --reload

访问：

http://127.0.0.1:8000/health
http://127.0.0.1:8000/docs

main:app 的含义是：从 main.py 文件中找到名为 app 的 FastAPI 对象。--reload 适合开发环境，代码变更自动重启；生产环境不要直接用 --reload。

请求参数怎么进入函数

Path 参数

Path 参数来自路径本身。

from fastapi import FastAPI

app = FastAPI()


@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

访问 /users/10 时，FastAPI 会把字符串 "10" 转成整数 10。如果访问 /users/abc，转换失败，会返回 422。

Query 参数

Query 参数来自 ?page=1&size=20。

@app.get("/users")
def list_users(page: int = 1, size: int = 20):
    return {"page": page, "size": size}

page 和 size 有默认值，所以调用方可以不传。生产项目中要限制 size，否则调用方一次查十万条会拖慢数据库和接口。

Body 参数

Body 常用于新增和修改。

from pydantic import BaseModel, Field


class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    age: int = Field(ge=0, le=150)


@app.post("/users")
def create_user(user: UserCreate):
    return {"name": user.name, "age": user.age}

这里 UserCreate 不是普通注释，它会参与运行时校验。调用方如果少传 name、age 不是数字、age 小于 0，FastAPI 会返回 422。

Header 参数

Header 常用于 Token、请求 ID、客户端版本。

from fastapi import Header


@app.get("/profile")
def profile(authorization: str | None = Header(default=None)):
    return {"authorization": authorization}

生产项目里不会直接返回 Token，而是解析 Token 得到用户身份。

422 是怎么来的

FastAPI 的参数校验大致过程：

flowchart TD
    A["请求进入路由"] --> B["读取 Path Query Header Body"]
    B --> C["根据函数签名和 Pydantic 模型转换类型"]
    C --> D{"类型和规则是否通过"}
    D -- "通过" --> E["调用接口函数"]
    D -- "失败" --> F["返回 422 校验错误"]

为什么不是 500？因为代码没有崩，服务端只是发现“客户端传来的数据不符合接口契约”。所以 422 是请求内容不合法，不是服务器内部错误。

如果业务规则不满足，例如“余额不足”“任务状态不能重复提交”，通常应该由业务代码返回 400 或自定义业务码，而不是依赖 Pydantic。

同步接口和异步接口

FastAPI 支持两种写法：

@app.get("/sync")
def sync_api():
    return {"mode": "sync"}


@app.get("/async")
async def async_api():
    return {"mode": "async"}

它们的区别不是“加了 async 就一定更快”，而是执行模型不同。

写法	适合场景	原理
def	同步数据库驱动、同步 SDK、普通业务代码	FastAPI 会放到线程池执行，避免阻塞事件循环
async def	异步 HTTP、异步数据库、WebSocket、流式响应	在事件循环中通过 await 释放等待时间

错误示例：

import time


@app.get("/bad")
async def bad_api():
    time.sleep(3)
    return {"ok": True}

time.sleep(3) 会阻塞事件循环。结果不是只有当前请求慢，而是同一个事件循环上的其他请求也会被拖慢。

正确写法：

import asyncio


@app.get("/good")
async def good_api():
    await asyncio.sleep(3)
    return {"ok": True}

如果必须调用同步阻塞函数：

import asyncio
import requests


def call_sync_service():
    response = requests.get("https://example.com", timeout=3)
    return response.status_code


@app.get("/proxy")
async def proxy():
    status_code = await asyncio.to_thread(call_sync_service)
    return {"status_code": status_code}

依赖注入

依赖注入用于把“多个接口都需要的逻辑”抽出来，例如登录用户、数据库会话、Service 对象、权限校验。

from fastapi import Depends, Header, HTTPException


def get_current_user(authorization: str | None = Header(default=None)):
    if authorization != "Bearer demo-token":
        raise HTTPException(status_code=401, detail="未登录")
    return {"id": 1, "name": "admin"}


@app.get("/me")
def me(current_user: dict = Depends(get_current_user)):
    return current_user

执行流程：

flowchart TD
    A["请求进入 /me"] --> B["发现 Depends"]
    B --> C["先执行 get_current_user"]
    C --> D{"是否通过鉴权"}
    D -- "否" --> E["直接返回 401"]
    D -- "是" --> F["把返回值注入 current_user"]
    F --> G["执行接口函数"]

依赖注入的价值是：接口函数只关心业务，不重复写鉴权、连接获取、公共参数解析。

中间件、依赖、异常处理怎么选

能力	执行范围	适合做什么	不适合做什么
中间件	几乎所有请求前后	请求日志、耗时统计、TraceId、CORS	复杂业务判断
依赖注入	指定接口或路由组	鉴权、权限、数据库会话、Service	全局响应包装
异常处理	出现异常时	统一错误格式、隐藏堆栈、记录错误	正常业务流程

请求前后日志中间件：

import time
import uuid
from fastapi import Request


@app.middleware("http")
async def access_log(request: Request, call_next):
    request_id = request.headers.get("X-Request-Id", str(uuid.uuid4()))
    start = time.perf_counter()
    response = await call_next(request)
    cost_ms = int((time.perf_counter() - start) * 1000)
    response.headers["X-Request-Id"] = request_id
    print(request_id, request.method, request.url.path, response.status_code, cost_ms)
    return response

全局异常处理

不要把 Python 堆栈直接返回给前端。生产项目应该把异常转换成统一结构，同时把详细堆栈记录到日志系统。

from fastapi import Request
from fastapi.responses import JSONResponse


class BizError(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message


@app.exception_handler(BizError)
async def biz_error_handler(request: Request, exc: BizError):
    return JSONResponse(
        status_code=400,
        content={"code": exc.code, "message": exc.message, "data": None},
    )


@app.exception_handler(Exception)
async def unknown_error_handler(request: Request, exc: Exception):
    print("unknown error", repr(exc))
    return JSONResponse(
        status_code=500,
        content={"code": "INTERNAL_ERROR", "message": "系统异常", "data": None},
    )

业务代码：

@app.post("/orders/{order_id}/pay")
def pay_order(order_id: int):
    if order_id <= 0:
        raise BizError("ORDER_ID_INVALID", "订单编号不合法")
    return {"code": "SUCCESS", "message": "success", "data": {"order_id": order_id}}

Lifespan 启动和关闭

商业项目启动时通常要初始化数据库连接池、HTTP 客户端、模型客户端、缓存等资源；关闭时要释放资源。

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.app_name = "asset-api"
    print("应用启动：初始化资源")
    yield
    print("应用关闭：释放资源")


app = FastAPI(lifespan=lifespan)

不要在每个请求里重复创建昂贵对象，例如数据库连接池、HTTP 客户端、向量数据库客户端。这样会导致连接数暴涨、延迟抖动、资源泄漏。

商业项目分层

推荐结构：

app/
  main.py
  api/
    asset_api.py
  schemas/
    asset_schema.py
  services/
    asset_service.py
  repositories/
    asset_repository.py
  core/
    config.py
    errors.py
    security.py

调用关系：

flowchart TD
    A["Router 接收 HTTP 请求"] --> B["Schema 校验输入"]
    B --> C["Dependency 注入用户和 Service"]
    C --> D["Service 编排业务规则"]
    D --> E["Repository 访问数据库"]
    D --> F["Client 调外部服务或 AI"]
    E --> G["数据库"]
    F --> H["外部系统"]
    D --> I["返回业务结果"]
    I --> J["Router 输出响应"]

职责划分：

层	负责	不应该负责
Router	HTTP 参数、状态码、响应模型	大段业务规则
Schema	输入输出结构和基础校验	查询数据库
Service	业务流程、事务边界、权限后的业务判断	解析 HTTP Header
Repository	SQL、ORM、数据持久化	业务编排
Client	调用第三方接口、AI、消息服务	保存主业务数据

完整分层 Demo

schemas/asset_schema.py：

from pydantic import BaseModel, Field


class AssetCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    source_system: str = Field(min_length=1, max_length=50)


class AssetVO(BaseModel):
    id: int
    name: str
    source_system: str

repositories/asset_repository.py：

class AssetRepository:
    def __init__(self):
        self._rows = []
        self._next_id = 1

    def save(self, name: str, source_system: str) -> dict:
        row = {
            "id": self._next_id,
            "name": name,
            "source_system": source_system,
        }
        self._next_id += 1
        self._rows.append(row)
        return row

    def list_all(self) -> list[dict]:
        return list(self._rows)

services/asset_service.py：

from repositories.asset_repository import AssetRepository


class AssetService:
    def __init__(self, repository: AssetRepository):
        self.repository = repository

    def create_asset(self, name: str, source_system: str) -> dict:
        if source_system not in {"HIS", "LIS", "PACS", "EMR"}:
            raise ValueError("来源系统不支持")
        return self.repository.save(name, source_system)

    def list_assets(self) -> list[dict]:
        return self.repository.list_all()

main.py：

from fastapi import Depends, FastAPI
from schemas.asset_schema import AssetCreate, AssetVO
from repositories.asset_repository import AssetRepository
from services.asset_service import AssetService

app = FastAPI()
repository = AssetRepository()


def get_asset_service():
    return AssetService(repository)


@app.post("/assets", response_model=AssetVO)
def create_asset(
    request: AssetCreate,
    service: AssetService = Depends(get_asset_service),
):
    return service.create_asset(request.name, request.source_system)


@app.get("/assets", response_model=list[AssetVO])
def list_assets(service: AssetService = Depends(get_asset_service)):
    return service.list_assets()

这不是为了“多建几个文件显得高级”，而是为了让 API 层、业务层、数据层可替换、可测试、可排查。

CORS、鉴权和 JWT 思路

前端项目和 API 不在同一个域名或端口时，浏览器会触发跨域限制。FastAPI 可以加 CORS 中间件：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

生产环境不要直接 allow_origins=["*"] 加 allow_credentials=True，否则容易把 Cookie 或凭证暴露给不可信来源。

JWT 鉴权基本链路：

flowchart TD
    A["用户登录"] --> B["服务端校验账号密码"]
    B --> C["签发 JWT"]
    C --> D["客户端保存 Token"]
    D --> E["请求携带 Authorization"]
    E --> F["服务端验证签名和过期时间"]
    F --> G["解析用户身份和权限"]

JWT 不是加密后的 Session，它主要是签名后的声明。不要在 JWT 中放密码、身份证号、密钥等敏感信息。

慢接口怎么排查

慢接口不要只说“加缓存”。要先定位慢在哪里。

flowchart TD
    A["发现接口慢"] --> B["看 access log 总耗时"]
    B --> C{"是否所有请求都慢"}
    C -- "是" --> D["看 CPU 内存 连接池 线程池"]
    C -- "否" --> E["按 request_id 查单次链路"]
    E --> F["拆分数据库 外部接口 AI 调用耗时"]
    F --> G{"数据库慢吗"}
    G -- "是" --> H["查 SQL 日志 EXPLAIN 索引 锁等待"]
    G -- "否" --> I{"外部接口或 AI 慢吗"}
    I -- "是" --> J["加超时 重试 降级 异步任务"]
    I -- "否" --> K["检查代码 CPU 计算和阻塞调用"]

建议每个请求至少记录：

字段	作用
request_id	串联一次请求的所有日志
method/path/status	定位接口
cost_ms	判断慢不慢
user_id	定位用户行为，注意脱敏
db_cost_ms	判断数据库是否慢
external_cost_ms	判断外部服务是否慢
error_code	统计错误类型

AI API 特别注意

AI 接口通常有三个特点：慢、贵、不稳定。不能像普通 CRUD 一样随便写。

问题	后果	建议
不设置超时	请求一直挂着，占用 worker	模型调用必须设置超时
不记录 token	成本不可控	记录输入输出 token、模型、用户
大任务同步等待	前端超时、网关超时	用任务 ID 轮询或流式响应
不做限流	被刷爆成本	按用户、IP、接口限流
不审计工具调用	AI 误删数据难追责	记录 prompt、工具、参数、结果

流式响应示例：

import asyncio
from fastapi.responses import StreamingResponse


async def fake_stream():
    for word in ["正在", "分析", "数据", "资产"]:
        yield f"data: {word}\n\n"
        await asyncio.sleep(0.3)


@app.get("/ai/chat/stream")
async def chat_stream():
    return StreamingResponse(fake_stream(), media_type="text/event-stream")

轮询式任务适合耗时更长的文件解析、批量入库、离线评估：

flowchart TD
    A["提交任务"] --> B["返回 task_id"]
    B --> C["后台执行 AI 或数据处理"]
    B --> D["客户端轮询任务状态"]
    C --> E["写入结果"]
    D --> F["查询到完成后读取结果"]

TestClient 测试 Demo

API 不是只能人工点页面，应该用测试保证核心行为。

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)


def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"


def test_create_asset_validation():
    response = client.post("/assets", json={"name": "", "source_system": "HIS"})
    assert response.status_code == 422

测试的意义不是追求覆盖率数字，而是保证参数校验、核心业务、异常处理和权限不会因为改代码被破坏。

部署基本思路

开发环境：

uvicorn main:app --reload

生产环境常见命令：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

注意：

1. workers 不是越多越好，太多会增加内存、数据库连接和上下文切换。
2. 多 worker 之间内存不共享，不要把重要状态只放在进程内字典里。
3. 数据库连接池大小要结合 worker 数一起算。
4. 生产要放在 Nginx、网关或容器平台后面统一做 HTTPS、限流、日志和健康检查。

健康检查接口：

@app.get("/health")
def health():
    return {"status": "UP"}

如果需要检查数据库，可以提供更严格的 /ready，但不要让健康检查做太重的操作。

商业场景：医疗数据资产接口

需求：外部采集程序把医院系统中的数据表登记到资产平台。

核心接口：

接口	作用
POST /assets	登记资产
GET /assets	分页查询资产
POST /collect-jobs	创建采集任务
GET /collect-jobs/{id}	查询任务状态

落地要点：

1. Router 只处理 HTTP，不直接写 SQL。
2. Service 校验来源系统、资产名称、任务状态。
3. Repository 保存资产、任务和审计日志。
4. 采集任务用异步任务或消息队列处理，不让 HTTP 一直等待。
5. 每次请求记录 request_id，便于排查“哪次采集失败”。
6. 对敏感字段做脱敏，避免日志泄露患者信息。

常见坑

问题	原因	后果	解决
async def 中调用同步阻塞库	不理解事件循环	并发请求全部变慢	换异步库或 asyncio.to_thread
Router 写满业务逻辑	没有分层	难测试、难复用	拆 Service/Repository
生产用 --reload	开发参数误用	不稳定、性能差	生产关闭 reload
进程内字典保存业务状态	多 worker 内存不共享	不同请求看到不同状态	状态放数据库/Redis
连接池过大	每个 worker 都建连接池	数据库连接被打满	结合 worker 数估算
不设置外部接口超时	请求长期挂起	worker 被耗尽	所有外部调用设置超时
异常返回堆栈	缺少异常处理	泄露内部信息	全局异常处理和日志

面试标准回答

FastAPI 一次请求怎么走？
客户端请求先到 Uvicorn，Uvicorn 按 ASGI 规范把请求交给 FastAPI。FastAPI 做路由匹配、参数提取、Pydantic 校验、依赖注入，然后执行接口函数。函数返回后，FastAPI 把结果序列化成 JSON，再由 Uvicorn 写回 HTTP 响应。

def 和 async def 接口怎么选？
同步阻塞代码、同步数据库驱动、普通 SDK 用 def 更稳；异步 HTTP、异步数据库、WebSocket、流式输出适合 async def。async def 里不能直接调用阻塞函数，否则会卡住事件循环，导致其他请求也变慢。

Web API 为什么要分层？
分层是为了把 HTTP 细节、业务规则、数据访问分开。Router 负责请求响应，Schema 负责校验，Service 负责编排业务，Repository 负责数据库。这样代码可测试、可复用、可排查，后续换数据库或改接口影响更小。

线上接口慢怎么排查？
先看 access log 和 request_id，确认是全局慢还是单接口慢；再拆数据库、外部接口、AI 调用、代码计算耗时。数据库慢看 SQL、EXPLAIN、索引、锁等待；外部接口慢加超时、重试、降级；代码慢检查阻塞调用、CPU 计算和连接池。

关联知识点

• Python 数据库访问
• Python 并发与异步
• Python 测试
• Python 日志与调试
• Python 与 AI 开发
• Python 面试题

小结

Python Web API 的核心不是背框架注解，而是理解请求链路、参数校验、依赖注入、异常处理、分层、日志和部署。真正的商业项目要关注可维护、可观测、可测试、可扩展：接口要能跑，更要能在出错、变慢、并发上来时说得清楚、查得到、改得动。