FastAPI 从入门到实战：为 refine 生态构建高性能 Python Web API-平芜编程栈

FastAPI 从入门到实战：为 refine 生态构建高性能 Python Web API

【免费下载链接】refineA React Framework for building internal tools, admin panels, dashboards & B2B apps with unmatched flexibility.项目地址: https://gitcode.com/GitHub_Trending/re/refine

本文来自 refine 项目官方博客中的后端教程（documentation/blog/2025-01-09-fast-api.md，2025 年 1 月 9 日更新，新增"错误处理"与"性能优化"两大章节），是一份完整的 FastAPI 入门与实战指南。读完本文，你将理解 FastAPI 的定位与优势、它与 Django/Flask/Pyramid 的差异，掌握环境搭建、路由创建、请求/响应体管理、自动文档、错误处理与性能优化等核心技能，并最终独立构建一个集数据库、认证与文件上传于一体的库存管理 REST API——这类 REST API 正好可以作为 refine 前端应用（通过 dataProvider 对接）的后端数据源。

FastAPI 是什么

FastAPI 是一个现代的 Python 微框架（microframework），旨在用 Python 语言简化 Web API 的创建过程。它允许开发者快速、轻松地构建 API，在保证性能与易管理性的同时，不牺牲代码质量和效率。

自进入后端开发领域以来，Python 凭借简洁与强大不断蚕食 PHP、.NET 等老牌技术的地盘；虽然其运行速度常被诟病，但它依然在该生态中蓬勃发展。Django、Flask 等框架相继出现，但底层速度问题始终存在。FastAPI 正是为解决这一痛点而被开发的：它一方面保留了 Flask 般的简洁风格，另一方面又开箱即用地提供了校验（validation）、文档（documentation）和响应编码（response encoding）等能力。

从技术栈上看，FastAPI 建立在Starlette之上，天然继承了 ASGI（Asynchronous Server Gateway Interface）生态与 async/await 异步能力；再叠加Pydantic的数据校验能力，构成了它的核心三件套。这些库的具体角色，会在后文的代码示例中逐一体现。

使用 FastAPI 的核心优势

性能（Performance）：FastAPI 通过充分利用 Pydantic 和 ASGI 生态等关键库与工具来最大化性能；同时，由于它牢固地构建在 Starlette 框架之上，能够无缝集成 async/await 异步功能。
可扩展性（Scalability）：FastAPI 的模块化与简洁性使其能够与负载均衡器无缝集成，从而方便水平扩展，并保证资源的有效利用。
自动文档（Automatic Documentation）：由于要求显式定义各种 API 组件，结合 Pydantic 的集成，FastAPI 能够自动生成 API 文档，默认提供 Swagger（OpenAPI）风格的交互式文档。
易用性（Ease-of-use）：FastAPI 是 Python 框架，因此继承了 Python 的全部优势；同时，创建服务器、构建端点都极其简单快捷。
请求校验（Request Validation）：FastAPI 借助 Pydantic 进行请求数据类型声明，能够提供更详细、对用户更友好的错误信息。

FastAPI 与其他 Python 框架的对比

Django vs FastAPI

Django 是一个功能丰富的 Python 后端框架，内置大量开箱即用的库，满足各类项目需求。它拥有强大的 ORM、认证机制和路由能力，适合开发复杂的 Web 应用。

FastAPI 则是一个刻意保持轻量的微框架。它虽然缺少庞大的内置库生态，却以"极快"作为补偿。与受限于"app 应用系统"的 Django 不同，FastAPI 使用现代的 Python 技术来释放其固有优势、提升性能。

Flask vs FastAPI

Flask 是 Python 开发者用来快速构建 Web 应用的轻量框架。它的设计理念是给开发者更多的控制权和灵活性，让应用结构完全贴合特定需求。

FastAPI 则专注于构建高性能、可扩展的应用，速度异常出色。它具备本文中讨论的诸多额外优势，更适合复杂应用的开发。

Pyramid vs FastAPI

Pyramid 遵循"只用你需要的"（use only what you need）哲学：提供一个极简核心，再通过各类附加组件与库进行扩充。这种模块化方式让开发者按需选配组件，得到一个轻量且高度可定制的框架。

FastAPI 则优先考虑开发者生产力与易用性：API 设计简单直观，文档清晰、示例丰富，还内置了自动文档生成等工具。

快速上手：环境准备与第一个路由

创建虚拟环境并安装依赖

像任何 Python 项目一样，首先需要创建虚拟环境（可参考 Python 官方 venv 文档），然后安装两个核心包——FastAPI与Uvicorn：

python -m pip install fastapi 'uvicorn[standard]'

fastapi：提供创建端点所需的方法与配置；
uvicorn：一个 ASGI 服务器，负责运行 FastAPI 应用。

[standard]附加选项会安装 Uvicorn 的推荐依赖集（如uvloop、httptools等），提升生产环境下的并发与性能表现。

编写第一个路由

创建一个main.py文件，它将包含你的全部代码。在文件中写入以下内容：

from fastapi import FastAPI fastapi = FastAPI() @fastapi.get("/") async def home(): return {"data": "Hello World"}

从fastapi模块导入FastAPI类后实例化它。这个实例可以像装饰器一样修饰处理函数来设置端点：它提供了 REST API 的各种动词（PUT、DELETE、PATCH、GET、POST）以及设置资源路径的方式。

启动开发服务器

在终端运行以下命令启动服务器：

uvicorn main:fastapi --reload

其中，main表示模块名（main.py），fastapi是FastAPI类的实例名。命令启动后，浏览器访问http://127.0.0.1:8000即可看到响应；--reload会在代码变更时自动重启服务，非常适合开发调试。

添加更多端点

要新增一个路由，只需创建一个处理函数，例如：

def handler(): return {"data": "from handler"}

然后使用之前创建的 FastAPI 实例，为函数添加装饰器：

@fastapi.post("/home-page")

这会将普通函数转换为一个 API 端点。同理，可以使用@fastapi.post、@fastapi.put、@fastapi.patch、@fastapi.delete创建不同类型的端点。

管理请求与响应体：FastAPI 的四种数据传递方式

向端点发送数据有多种方式，FastAPI 对它们的处理各有侧重。

路径参数（Path Parameters）

将短数据直接附加在 URL 路径中：

# other data goes here @fastapi.get("/{name}") async def get_name(name: str): return {"name": name}

上面的name是一个路径参数，从 URL 中提取后作为参数传给get_name函数，方便你直接访问 URL 路径中携带的数据。

查询参数（Query Parameters）

查询参数与路径参数类似，区别在于它们以问号（?）为前缀追加在 URL 末尾：

@fastapi.get("/") async def get_api_data(data_type: str, skip: int = 0, limit: int = 10): return {"data_type": data_type, "skip": skip, "limit": limit}

这里skip和limit是查询参数，通过?key=value的格式提供。skip与limit的默认值分别为 0 和 10，可以在 URL 中传入不同值进行覆盖。

请求体参数（Body Parameters）

与前两者不同，请求体参数把数据编码后附加到发往端点的请求中，常用于 POST/PATCH 等写入操作：

from pydantic import BaseModel from fastapi import FastAPI fastapi = FastAPI() class Item(BaseModel): name: str price: float @fastapi.post("/items") async def create_item(item: Item): return {"item": item}

示例中定义了一个继承自 PydanticBaseModel的Item类，它完整定义了请求体的参数结构，并向 FastAPI 提供了更多上下文（如类型、必填性、默认值），校验与自动文档都以此为基础。

请求头与 Cookie（Headers and Cookies）

请求头与 Cookie 是向服务器传递上下文信息的常用手段。

Headers：

from fastapi import FastAPI, Header app = FastAPI() @app.get("/items") async def read_items(user_agent: str = Header(None)): return {"User-Agent": user_agent}

user_agent参数使用默认值Header(None)，告诉 FastAPI 从请求中提取该请求头的值：若请求中存在该头，则赋给user_agent；否则为None。

Cookies：

from fastapi import FastAPI, Cookie app = FastAPI() @app.get("/items/") async def read_items(session_token: str = Cookie(None)): return {"session_token": session_token}

与请求头同理，session_token参数使用默认值Cookie(None)，FastAPI 会把请求中携带的 Cookie 注入该参数；若无 Cookie，则为None。

预览自动生成的 API 文档

服务器运行后，在浏览器访问http://127.0.0.1:8000/docs，即可看到 Swagger UI 交互式文档。它会自动列出所有端点、请求/响应模型、参数说明，并支持直接在页面上发送请求测试——这正是"自动文档"优势的直观体现，也是开发阶段调试 API 的得力工具。

实战：构建库存管理 REST API

为了展示 FastAPI 的真实能力，我们将构建一个假设的库存管理（inventory）应用的 REST API。这个 API 将连接数据库、支持图片上传、并为部分路由添加保护。端点设计如下：

GET /items—— 获取服务器上所有条目
GET /items/{item_id}—— 获取指定条目
POST /items—— 新增一个条目
PATCH /items/{item_id}—— 更新指定条目
DELETE /items/{item_id}—— 从库存中删除条目

在此基础上，还会创建处理文件服务（serving files）的端点，并使用中间件为部分端点添加认证。

前置条件

具备 Python 基础知识；
了解 HTTP、JSON、REST API 以及 Python 虚拟环境；
有一个可用的终端；
安装 Python 3.10。

设置项目结构

创建虚拟环境后，在虚拟环境目录中创建src目录，代码将存放于此，并用代码编辑器打开该目录。项目最终将包含以下文件：

src/ ├── main.py # 应用入口，定义全部端点 ├── utils.py # 工具函数（如 find_item） ├── middleware.py # 认证中间件 └── database.py # 数据库引擎、会话与模型

安装依赖

除了 Uvicorn 和 FastAPI 这两个核心依赖外，由于需要增强服务器的能力（数据库连接 + 文件上传系统），还需安装databases和SQLAlchemy：

pip install 'fastapi[all]' 'uvicorn[standard]' databases sqlalchemy

fastapi[all]会安装 FastAPI 的全部可选依赖，方便一次性获得完整的开发体验。

创建核心端点

在src目录中创建main.py（应用入口），写入以下内容：

from fastapi import FastAPI, HTTPException, Depends from .utils import find_item from .middleware import authenticate from pydantic import BaseModel, Field from typing import Optional fastapi = FastAPI() inventory = [ {"id": 1, "name": "Treasure", "quantity": 3} ] class Item(BaseModel): name: str quantity: int class ItemUpdate(BaseModel): name: Optional[str] = Field(None, description="Optional name of the item") quantity: Optional[int] = Field(None, description="Optional quantity of the item") @fastapi.get("/items") async def get_items(): return {"items": inventory} @fastapi.get("/items/{item_id}") async def get_item(item_id: int): item, idx = find_item(inventory, lambda x: x["id"] == item_id) return {"item": item} @fastapi.delete("/items/{item_id}") async def delete_item(item_id: int, authenticated: bool = Depends(authenticate)): item, idx = find_item(inventory, lambda x: x["id"] == item_id) if idx == -1: raise HTTPException(status_code=404, detail="Item not found") inventory.pop(idx) return {"item": item} @fastapi.post("/items") async def add_item(data: Item): item = { "id": len(inventory) + 1, "name": data.name, "quantity": data.quantity } inventory.append(item) return item @fastapi.patch("/items/{item_id}") async def update_item(item_id: int, item_update: ItemUpdate): item, idx = find_item(inventory, lambda x: x["id"] == item_id) if idx == -1: raise HTTPException(status_code=404, detail="Item not found") if item_update.name is not None: item["name"] = item_update.name if item_update.quantity is not None: item["quantity"] = item_update.quantity inventory[idx] = item return item

这段代码包含五个端点处理函数，并用一个本地inventory数据存储保存所有条目。ItemUpdate类定义了 PATCH 端点的请求体参数，允许字段可选：通过typing模块的Optional与 Pydantic 的Field（可附带描述信息）实现可选字段。

几点说明：

find_item(collection, predicate)是从 utils.py 导入的工具函数，接收一个列表和匹配谓词，返回(item, index)元组；当未找到匹配项时索引为 -1。请把它实现后放入src/utils.py；
authenticate依赖来自后文"认证中间件"小节创建的middleware.py（原文中该端点即引用了这一依赖，本文将所需导入一并补齐）;
原文档在delete_item中使用了return HTTPException(...)，正确做法是像update_item那样raise HTTPException(status_code=404, detail="Item not found")，本文已按正确方式给出。

完成后即可运行服务器，并通过/docs文档页测试这些端点。

FastAPI 错误处理

FastAPI 的错误处理直观且简单：它内置了HTTPException类，可以在出错时轻松返回规范的错误响应。

处理无效输入（Invalid IDs）

from fastapi import FastAPI, HTTPException app = FastAPI() @app.get("/items/{item_id}") async def read_item(item_id: int): if item_id <= 0: # Raise an error if the item_id is invalid raise HTTPException( status_code=400, detail="Invalid ID. ID must be greater than 0." ) return {"item_id": item_id}

当用户请求的item_id小于等于 0 时，响应如下：

{ "detail": "Invalid ID. ID must be greater than 0." }

自定义错误响应

你还可以让错误响应携带更多信息，例如把detail参数指定为字典：

@app.get("/users/{user_id}") async def read_user(user_id: int): if user_id > 100: raise HTTPException( status_code=404, detail={'error': 'User not found', 'user_id': user_id} ) return {"user_id": user_id}

无效 user_id 的响应可能长这样：

{ "error": "User not found", "user_id": 150 }

捕获服务器意外错误

对于意料之外的异常，可以使用全局异常处理器：

from fastapi import Request from fastapi.responses import JSONResponse @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): return JSONResponse( status_code=500, content={"message": "An unexpected error occurred. Please try again."} )

这样即使服务器内部出错，用户也会得到友好的错误提示，而不是堆栈信息。

优化 FastAPI 性能

FastAPI 默认就很快，但仍有一些做法可以进一步提升速度。

使用异步库

使用httpx这类异步库发起非阻塞的 HTTP 请求：

import httpx from fastapi import FastAPI app = FastAPI() @app.get("/data") async def get_data(): async with httpx.AsyncClient() as client: response = await client.get("https://api.example.com/data") return response.json()

在 async 端点内使用async with+await，让 I/O 等待期间让出事件循环，从而提升整体吞吐。

实现缓存

缓存能减少服务器对数据库或外部 API 的重复调用，例如使用 Redis：

import aioredis from fastapi import FastAPI app = FastAPI() redis = aioredis.from_url("redis://localhost") @app.get("/items/{item_id}") async def read_item(item_id: int): # Check if item is cached cached_item = await redis.get(f"item:{item_id}") if cached_item: return {"item": cached_item.decode("utf-8")} # Simulate database fetch item = f"Item {item_id}" await redis.set(f"item:{item_id}", item) return {"item": item}

命中缓存时直接返回，避免重复计算与磁盘 I/O。

使用负载均衡

配置 Nginx 或 Traefik 等负载均衡器来承接请求压力。一个典型的 Nginx 配置示例：

server { listen 80; location / { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_set_header Host $host; } }

优化查询性能

为数据库建立合适的索引，并避免获取非必要的数据：

from sqlalchemy.orm import Session from fastapi import Depends @app.get("/users") async def get_users(limit: int = 10, db: Session = Depends(get_db)): return db.query(User).limit(limit).all()

通过limit限制返回行数，配合索引可显著降低响应延迟。

启用 Gzip 压缩

压缩响应体积可以减少传输时间：

pip install fastapi-compression

然后在应用中挂载压缩中间件：

from fastapi import FastAPI from fastapi_compression import CompressionMiddleware app = FastAPI() app.add_middleware(CompressionMiddleware)

综合运用以上技巧，你的 FastAPI 应用将既快又稳，足以应对高负载场景。

进阶：认证中间件、数据库集成与文件上传

真实世界的 API 往往比库存 API 更复杂：你可能需要持久化数据、在请求前校验凭据，或处理文件。下面为库存应用依次加入认证中间件、数据库与文件上传。

实现认证中间件

中间件好比 API 世界的"阀门"，可用于限制特定用户访问、为请求附加上下文等。这里创建一个中间件，只允许持有特定凭据的客户端访问部分端点。

在src目录新建middleware.py：

from fastapi import HTTPException, Depends from fastapi.security import HTTPBasic, HTTPBasicCredentials security = HTTPBasic() def authenticate(credentials: HTTPBasicCredentials = Depends(security)): correct_username = 'admin' correct_password = 'password' if credentials.username != correct_username or credentials.password != correct_password: raise HTTPException(status_code=401, detail="Unauthorized") return True

authenticate函数实现了 HTTP Basic 认证。要把它挂到指定路由上，回到main.py，在目标端点参数中加入：

authenticated: bool = Depends(authenticate)

之后处理函数长这样：

@fastapi.post("/items") async def add_item(data: Item, authenticated: bool = Depends(authenticate)): item = { "name": data.name, "id": len(inventory) + 1, "quantity": data.quantity } inventory.append(item) return item

记得从middleware.py导入authenticate、从fastapi导入Depends。此后，未通过 Basic 认证的请求访问该端点将得到 401 响应。

数据库集成

首先确保已安装sqlalchemy与databases。然后在src目录创建database.py，导入所需组件：

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

出于简单考虑，这里使用 SQLite。创建数据库引擎：

DATABASE_URL = "sqlite:///./database.db" engine = create_engine(DATABASE_URL)

这是数据库的基础连接配置。接着创建SessionLocal（调用时创建数据库会话）与Base（所有数据库模型的基类）：

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

然后定义一个数据库模型：

class DBItem(Base): __tablename__ = "items" id = Column(Integer, primary_key=True, index=True) quantity = Column(Integer) name = Column(String)

接下来把DBItem和SessionLocal导入main.py，并更新路由处理器：

@fastapi.post("/items") def create_item(item: Item, authenticated: bool = Depends(authenticate)): db = SessionLocal() new_item = DBItem(name=item.name, quantity=item.quantity) db.add(new_item) db.commit() db.refresh(new_item) return {"item": new_item} @fastapi.get("/items") async def get_items(): db = SessionLocal() items = db.query(DBItem).all() return {"items": items} @fastapi.get("/items/{item_id}") def get_item(item_id: int): db = SessionLocal() item = db.query(DBItem).filter(DBItem.id == item_id).first() if not item: raise HTTPException(status_code=404, detail="Item not found") return {"item": item} @fastapi.patch("/items/{item_id}") def update_item(item_id: int, item: ItemUpdate, authenticated: bool = Depends(authenticate)): db = SessionLocal() db_item = db.query(DBItem).filter(DBItem.id == item_id).first() if not db_item: raise HTTPException(status_code=404, detail="Item not found") db_item.name = item.name db_item.quantity = item.quantity db.commit() db.refresh(db_item) return {"item": db_item} @fastapi.delete("/items/{item_id}") def delete_item(item_id: int, authenticated: bool = Depends(authenticate)): db = SessionLocal() db_item = db.query(DBItem).filter(DBItem.id == item_id).first() if not db_item: raise HTTPException(status_code=404, detail="Item not found") db.delete(db_item) db.commit() return {"message": "Item deleted"}

可以看到，每个端点都会实例化SessionLocal来执行数据库查询；查询完成后通过commit()持久化到数据库。PATCH 端点更新时注意对item.name与item.quantity做空值判断（可参考前面ItemUpdate的可选字段设计）。

文件上传与静态文件服务

FastAPI 让文件上传变得非常简单。首先更新database.py中的DBItem，添加图片字段：

class DBItem(Base): # other database fields goes here image_src = Column(String)

然后回到main.py，从fastapi导入File与UploadFile，导入os模块，并从fastapi.responses（即 Starlette 的 responses）导入FileResponse，接着添加以下端点：

@fastapi.patch("/item-image/{item_id}") async def upload_file(item_id: int, file: UploadFile = File()): db = SessionLocal() db_item = db.query(DBItem).filter(DBItem.id == item_id).first() if not db_item: raise HTTPException(status_code=404, detail="Item not found") file_path = os.path.join("uploads", file.filename) with open(file_path, "wb") as f: f.write(await file.read()) db_item.image_src = file.filename db.commit() db.refresh(db_item) return {"item": db_item} @fastapi.get("/static/{file}") async def serve_file(file: str): return FileResponse(os.path.join("uploads", file))

第一个路由更新指定条目的image_src字段并将文件写入服务器；第二个路由serve_file负责从服务器取回文件。Starlette 还提供了多种响应类型，FileResponse是其中之一，可直接用于路由处理器。

至此，你的库存管理 API 已经具备完整的增删改查、Basic 认证、SQLite 持久化与文件上传/服务能力，可以在 Swagger 文档页（/docs）中逐一验证。

对比一览表

Feature	FastAPI	Django	Flask	Pyramid
Performance	High (ASGI & async/await)	Moderate	Moderate	Moderate
Auto Documentation	Yes (Swagger/OpenAPI)	No	No	No
Database Integration	Requires external libraries	Built-in ORM (Django ORM)	Requires external libraries	Requires external libraries
Scalability	High	High	Moderate	High
Learning Curve	Easy	Moderate	Easy	Moderate

结语

至此，你应该已经对使用 FastAPI 构建后端应用有了相当深入的认识。FastAPI 是一个出色、易用且高效的 API 开发工具：它兼具微框架的灵活性，又能交出卓越的性能表现，是 API 开发需求的绝佳选择。

在 refine 生态中，前端应用通过 dataProvider 与后端 API 交互（参见数据获取教程，refine 也提供了现成的 REST 数据提供者包），因此本文构建的 FastAPI REST API 可以非常自然地成为 refine 管理后台/内部工具的数据源。如果你想继续深入，可以参考 FastAPI 官方文档、Starlette 文档与 Pydantic 文档了解更高级的用法。

【免费下载链接】refineA React Framework for building internal tools, admin panels, dashboards & B2B apps with unmatched flexibility.项目地址: https://gitcode.com/GitHub_Trending/re/refine

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考