问题与目标
把 Python 函数变成可调用服务,不只是监听一个端口。服务还要明确输入格式、校验规则、状态码和失败响应。本篇用 FastAPI 实现内存版任务接口,先聚焦 HTTP 边界。
完成标准:能启动 ASGI 服务,创建和查询任务,观察自动校验,并区分框架、ASGI 服务器与业务函数的职责。
核心概念
FastAPI 负责路由、数据校验和响应转换;Uvicorn 是运行 ASGI 应用的服务器;路由函数承载应用入口,但复杂业务逻辑应继续下沉到独立模块。
路径参数用于定位资源,请求体传递结构化数据,查询参数适合筛选和分页。创建成功返回 201,不存在返回 404,输入不符合模型时框架返回 422。可预期业务错误应转换为稳定的 HTTP 响应,未知错误留给统一日志记录。
先把契约写清楚,再写路由:
| 方法与路径 | 输入 | 成功响应 | 常见失败 |
|---|---|---|---|
POST /tasks | JSON 请求体 | 201 和任务 JSON | 422 输入无效 |
GET /tasks?limit=20 | 查询参数 | 200 和数组 | 422 limit 越界 |
GET /tasks/{id} | 路径参数 | 200 和任务 JSON | 404 不存在 |
响应模型不仅生成文档,也限制对外字段。数据库记录中即使包含内部备注或审计字段,也不应在没有设计的情况下自动暴露。
可运行实现
bash
python -m pip install fastapi uvicorn
保存为 api.py:
python
from typing import Literal
from fastapi import FastAPI, HTTPException, Query, status
from pydantic import BaseModel, Field
app = FastAPI(title="Task API")
class TaskCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
class Task(BaseModel):
id: int
title: str
status: Literal["todo", "doing", "done"]
tasks: dict[int, Task] = {}
@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate) -> Task:
task_id = max(tasks, default=0) + 1
task = Task(id=task_id, title=payload.title, status="todo")
tasks[task_id] = task
return task
@app.get("/tasks", response_model=list[Task])
def list_tasks(limit: int = Query(default=20, ge=1, le=100)) -> list[Task]:
return list(tasks.values())[:limit]
@app.get("/tasks/{task_id}", response_model=Task)
def get_task(task_id: int) -> Task:
task = tasks.get(task_id)
if task is None:
raise HTTPException(status_code=404, detail="task not found")
return task
启动与验收:
bash
uvicorn api:app --reload --host 127.0.0.1 --port 8000
curl -i -X POST http://127.0.0.1:8000/tasks \
-H 'Content-Type: application/json' \
-d '{"title":"build API boundary"}'
curl -i http://127.0.0.1:8000/tasks/1
curl -i http://127.0.0.1:8000/tasks/999
--reload 只用于开发。内存字典会在重启后丢失数据,也不适合多进程,这正是下一篇接入数据库的原因。
接口契约可以不启动真实端口直接测试:
python
from fastapi.testclient import TestClient
from api import app
client = TestClient(app)
def test_create_and_get_task() -> None:
created = client.post("/tasks", json={"title": "test contract"})
assert created.status_code == 201
task_id = created.json()["id"]
fetched = client.get(f"/tasks/{task_id}")
assert fetched.status_code == 200
assert fetched.json()["title"] == "test contract"
def test_reject_empty_title() -> None:
response = client.post("/tasks", json={"title": ""})
assert response.status_code == 422
运行 pytest -q,同时验证成功路径和输入边界。测试进程中的内存状态会共享,因此更完整的测试应在每个用例前重置存储。
常见问题与排查
Could not import module:确认当前目录、文件名和api:app中变量名。- 返回 422:查看响应中的字段位置和规则,它说明请求已到达但输入未通过校验。
- 路由顺序冲突:固定路径和动态路径设计应避免歧义,例如
/tasks/stats与/tasks/{task_id}。 - 在普通
def与async def间盲目切换:同步数据库驱动不会因外层改成异步函数而非阻塞。 - 把内部异常文本原样返回:可能泄露 SQL、路径或密钥;对外给稳定错误,对内记录完整堆栈。
- 只测试 200:创建接口的 201、输入错误的 422 和资源不存在的 404 都属于契约的一部分。
小结
Web API 的核心是稳定契约:方法和路径表达操作,模型定义输入输出,状态码表达结果。框架帮助守住协议边界,业务正确性和数据持久化仍需独立设计。
许可协议:CC BY-NC 4.0
更新于 2 小时前
觉得文章有帮助?点个赞吧!
0 条评论


