探索FastAPI构建高性能Web应用程序的完整指南从入门到实践掌握现代Python框架的核心技巧

引言

在现代Web开发领域，高性能、易用性和可扩展性是开发者们不断追求的目标。Python作为一种流行的编程语言，在Web开发领域有着强大的影响力。而在众多的Python Web框架中，FastAPI凭借其快速、现代和易用的特性，正迅速成为开发者们的首选之一。本文将深入探讨FastAPI框架，从基础概念到高级应用，帮助读者全面掌握这一现代Python框架的核心技巧，从而能够构建高性能的Web应用程序。

FastAPI概述

FastAPI是一个现代、快速（高性能）的Web框架，用于构建API。它基于Python 3.7+的类型提示功能，提供了自动化的数据验证、序列化和文档生成。FastAPI的设计目标是提高开发效率和性能，同时保持代码的简洁和可维护性。

FastAPI的核心特性

高性能：基于Starlette框架，FastAPI提供了极高的性能，接近于Node.js和Go。这使得它成为构建高并发应用的理想选择。
自动文档生成：内置支持Swagger UI和ReDoc，自动生成API文档。这使得开发者可以轻松地与前端团队协作，也便于API的测试和维护。
数据验证：基于Pydantic，提供强大的数据验证和序列化功能。这确保了数据的完整性和安全性，同时减少了手动验证的工作量。
异步支持：完全支持异步编程，适合处理高并发请求。通过Python的async和await语法，开发者可以轻松编写非阻塞代码。
类型提示：利用Python的类型提示功能，提高代码的可读性和可维护性。这不仅有助于开发过程中的错误检测，还能提供更好的IDE支持。

为什么选择FastAPI？

在众多Python Web框架中，FastAPI之所以脱颖而出，主要归功于以下几个优点：

开发效率：FastAPI的简洁语法和自动化功能大大提高了开发效率。开发者可以用更少的代码完成更多的工作。
性能优越：在高并发场景下，FastAPI的性能表现优异，可以与Node.js和Go相媲美。
文档友好：自动生成的API文档使得前后端协作更加顺畅，也便于API的测试和维护。
社区支持：FastAPI拥有活跃的社区和丰富的插件生态，为开发者提供了大量的资源和支持。

环境搭建

在开始使用FastAPI之前，我们需要进行一些环境准备工作。

安装Python

首先，确保你的系统中已经安装了Python 3.7或更高版本。你可以通过在终端中运行以下命令来检查Python版本：

python --version

或者：

python3 --version

创建虚拟环境

为了保持项目的依赖隔离，建议创建一个虚拟环境：

# 创建项目目录 mkdir fastapi_project cd fastapi_project # 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Windows venvScriptsactivate # macOS/Linux source venv/bin/activate

安装FastAPI和Uvicorn

接下来，安装FastAPI和Uvicorn（一个ASGI服务器，用于运行FastAPI应用）：

pip install fastapi uvicorn

或者，你可以安装包含所有额外依赖的FastAPI版本：

pip install fastapi[all]

这将安装FastAPI及其所有可选依赖，包括uvicorn、python-multipart（用于表单数据）、python-jose（用于JWT令牌）等。

基础入门：创建第一个FastAPI应用

现在，让我们创建一个简单的FastAPI应用程序。

创建基本应用

创建一个名为main.py的文件，并添加以下代码：

from fastapi import FastAPI # 创建FastAPI应用实例 app = FastAPI() # 定义根路径的路由和处理函数 @app.get("/") async def root(): return {"message": "Hello World"}

运行应用

在终端中，使用以下命令启动应用：

uvicorn main:app --reload

其中：

main：是Python文件名（main.py）
app：是在main.py中创建的FastAPI实例
--reload：使服务器在代码更改后自动重启

测试应用

打开浏览器，访问http://127.0.0.1:8000，你将看到JSON响应：

{"message": "Hello World"}

访问自动生成的文档

FastAPI自动生成交互式API文档。访问http://127.0.0.1:8000/docs，你将看到Swagger UI界面：

探索FastAPI构建高性能Web应用程序的完整指南从入门到实践掌握现代Python框架的核心技巧-小辉娱乐网

或者，访问http://127.0.0.1:8000/redoc，查看ReDoc文档：

核心概念详解

路径操作

在FastAPI中，”路径操作”是指定义如何处理特定HTTP方法和URL路径的组合。FastAPI使用装饰器来定义路径操作。

路径参数

路径参数是URL路径的一部分，用于捕获动态值。例如：

from fastapi import FastAPI app = FastAPI() @app.get("/items/{item_id}") async def read_item(item_id: int): return {"item_id": item_id}

在这个例子中，item_id是一个路径参数。FastAPI会将其转换为整数（因为我们在函数参数中指定了类型为int）。

查询参数

查询参数是URL中?后面的键值对。例如：

from fastapi import FastAPI app = FastAPI() @app.get("/items/") async def read_item(skip: int = 0, limit: int = 10): return {"skip": skip, "limit": limit}

在这个例子中，skip和limit是查询参数，它们都有默认值。

请求体

当需要发送复杂的数据结构时，可以使用请求体。FastAPI使用Pydantic模型来声明请求体：

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Item(BaseModel): name: str description: str | None = None price: float tax: float | None = None @app.post("/items/") async def create_item(item: Item): return item

在这个例子中，我们定义了一个Item模型，它有四个字段：name（必需的字符串）、description（可选的字符串）、price（必需的浮点数）和tax（可选的浮点数）。

数据验证与序列化

FastAPI使用Pydantic进行数据验证和序列化。Pydantic模型不仅用于声明请求体，还用于验证和转换输入数据。

数据验证示例

from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app = FastAPI() class Item(BaseModel): name: str = Field(..., min_length=1, max_length=50) description: str | None = Field(None, max_length=300) price: float = Field(..., gt=0) tax: float | None = Field(None, ge=0) @app.post("/items/") async def create_item(item: Item): if item.name == "admin": raise HTTPException(status_code=400, detail="Item name cannot be 'admin'") return item

在这个例子中，我们使用Field类为每个字段添加了验证规则：

name：必填，长度在1到50之间
description：可选，最大长度为300
price：必填，必须大于0
tax：可选，必须大于或等于0

如果输入数据不符合这些规则，FastAPI会自动返回一个详细的错误响应。

响应模型

你可以指定响应模型，以确保输出数据符合特定的结构：

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Item(BaseModel): name: str description: str | None = None price: float tax: float = 10.5 class ResponseItem(BaseModel): name: str description: str | None = None price_with_tax: float @app.post("/items/", response_model=ResponseItem) async def create_item(item: Item): price_with_tax = item.price + item.tax return {"name": item.name, "description": item.description, "price_with_tax": price_with_tax}

在这个例子中，我们定义了一个ResponseItem模型作为响应模型。FastAPI会确保返回的数据符合这个模型的结构。

进阶特性

异步编程

FastAPI完全支持异步编程，这使得它能够高效地处理大量并发请求。使用Python的async和await语法，你可以轻松编写非阻塞代码。

异步路径操作函数

import asyncio from fastapi import FastAPI app = FastAPI() @app.get("/") async def read_root(): await asyncio.sleep(1) # 模拟耗时操作 return {"message": "Hello World"}

异步数据库操作

import asyncio from databases import Database from fastapi import FastAPI app = FastAPI() database = Database("sqlite:///test.db") @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() @app.get("/items/{item_id}") async def read_item(item_id: int): query = "SELECT * FROM items WHERE id = :id" return await database.fetch_one(query, {"id": item_id})

依赖注入系统

FastAPI有一个强大的依赖注入系统，允许你轻松地集成组件、数据库连接、认证等。

简单依赖

from fastapi import FastAPI, Depends app = FastAPI() async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100): return {"q": q, "skip": skip, "limit": limit} @app.get("/users/") async def read_users(commons: dict = Depends(common_parameters)): return commons @app.get("/items/") async def read_items(commons: dict = Depends(common_parameters)): return commons

在这个例子中，common_parameters函数是一个依赖，它被read_users和read_items路径操作函数共享。

类作为依赖

from fastapi import FastAPI, Depends app = FastAPI() class CommonQueryParams: def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100): self.q = q self.skip = skip self.limit = limit @app.get("/users/") async def read_users(commons: CommonQueryParams = Depends()): return commons

在这个例子中，CommonQueryParams类被用作依赖。FastAPI会处理类的实例化，并将其注入到路径操作函数中。

中间件

中间件是在每个请求被特定路径操作处理之前和之后运行的函数。FastAPI允许你添加中间件来处理跨域资源共享、CORS、请求日志等。

添加中间件

from fastapi import FastAPI, Request import time app = FastAPI() @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 import FastAPI from fastapi.middleware.cors import CORSMiddleware app = FastAPI() origins = [ "http://localhost", "http://localhost:8080", ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )

在这个例子中，我们添加了CORS中间件，允许来自特定来源的跨域请求。

异常处理

FastAPI提供了内置的异常处理，你也可以自定义异常处理器。

抛出HTTP异常

from fastapi import FastAPI, HTTPException app = FastAPI() items = {"foo": "The Foo Wrestlers"} @app.get("/items/{item_id}") async def read_item(item_id: str): if item_id not in items: raise HTTPException(status_code=404, detail="Item not found") return {"item": items[item_id]}

自定义异常处理器

from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError app = FastAPI() @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): return JSONResponse( status_code=422, content={"detail": exc.errors(), "body": exc.body}, )

在这个例子中，我们自定义了一个验证异常处理器，它返回更详细的错误信息。

数据库集成

FastAPI可以与各种数据库集成，包括SQLAlchemy、ORM、MongoDB等。下面我们以SQLAlchemy为例，介绍如何在FastAPI中集成数据库。

安装依赖

首先，安装SQLAlchemy和数据库驱动：

pip install sqlalchemy databases

配置数据库

from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from databases import Database # SQLAlchemy配置 SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() # databases配置 DATABASE_URL = "sqlite:///./test.db" database = Database(DATABASE_URL) # 定义模型 class Item(Base): __tablename__ = "items" id = Column(Integer, primary_key=True, index=True) title = Column(String, index=True) description = Column(String, index=True) # 创建表 Base.metadata.create_all(bind=engine)

CRUD操作

from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from pydantic import BaseModel from typing import List app = FastAPI() # Pydantic模型 class ItemBase(BaseModel): title: str description: str | None = None class ItemCreate(ItemBase): pass class Item(ItemBase): id: int class Config: orm_mode = True # 依赖项 def get_db(): db = SessionLocal() try: yield db finally: db.close() # CRUD操作 @app.post("/items/", response_model=Item) async def create_item(item: ItemCreate, db: Session = Depends(get_db)): db_item = Item(**item.dict()) db.add(db_item) db.commit() db.refresh(db_item) return db_item @app.get("/items/", response_model=List[Item]) async def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): items = db.query(Item).offset(skip).limit(limit).all() return items @app.get("/items/{item_id}", response_model=Item) async def read_item(item_id: int, db: Session = Depends(get_db)): db_item = db.query(Item).filter(Item.id == item_id).first() if db_item is None: raise HTTPException(status_code=404, detail="Item not found") return db_item

异步数据库操作

from fastapi import FastAPI, HTTPException from databases import Database from pydantic import BaseModel from typing import List app = FastAPI() database = Database("sqlite:///./test.db") # Pydantic模型 class ItemBase(BaseModel): title: str description: str | None = None class ItemCreate(ItemBase): pass class Item(ItemBase): id: int class Config: orm_mode = True # 启动和关闭事件 @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() # CRUD操作 @app.post("/items/", response_model=Item) async def create_item(item: ItemCreate): query = "INSERT INTO items (title, description) VALUES (:title, :description)" values = {"title": item.title, "description": item.description} last_record_id = await database.execute(query, values) return {**item.dict(), "id": last_record_id} @app.get("/items/", response_model=List[Item]) async def read_items(skip: int = 0, limit: int = 100): query = "SELECT * FROM items ORDER BY id LIMIT :limit OFFSET :skip" items = await database.fetch_all(query, {"limit": limit, "skip": skip}) return items @app.get("/items/{item_id}", response_model=Item) async def read_item(item_id: int): query = "SELECT * FROM items WHERE id = :id" item = await database.fetch_one(query, {"id": item_id}) if item is None: raise HTTPException(status_code=404, detail="Item not found") return item

性能优化

FastAPI本身已经是一个高性能框架，但我们还可以采取一些措施来进一步优化应用性能。

异步数据库操作

使用异步数据库驱动可以显著提高性能，特别是在高并发场景下。如上所述，你可以使用databases库与SQLAlchemy集成，实现异步数据库操作。

缓存

缓存是提高应用性能的有效方法。你可以使用Redis或Memcached等缓存系统来缓存频繁访问的数据。

from fastapi import FastAPI, Depends from fastapi_cache import FastAPICache, Coder from fastapi_cache.backends.redis import RedisBackend from fastapi_cache.decorator import cache from redis import asyncio as aioredis from pydantic import BaseModel app = FastAPI() @app.on_event("startup") async def startup(): redis = aioredis.from_url("redis://localhost") FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache") class Item(BaseModel): name: str description: str | None = None price: float tax: float | None = None @app.get("/items/{item_id}") @cache(expire=60) async def read_item(item_id: int): # 模拟耗时操作 import time time.sleep(2) return {"item_id": item_id, "name": "Example Item"}

在这个例子中，我们使用fastapi-cache库和Redis来实现缓存。@cache装饰器会缓存函数的结果，指定的时间（这里是60秒）内再次调用该函数将直接返回缓存的结果。

响应压缩

启用响应压缩可以减少网络传输的数据量，从而提高性能。

from fastapi import FastAPI from fastapi.middleware.gzip import GZipMiddleware app = FastAPI() app.add_middleware(GZipMiddleware, minimum_size=1000)

在这个例子中，我们添加了GZip中间件，它会在响应大小超过1000字节时自动压缩响应。

静态文件处理

对于静态文件（如CSS、JavaScript和图像），使用专门的静态文件服务器可以提高性能。

from fastapi import FastAPI from fastapi.staticfiles import StaticFiles app = FastAPI() # 挂载静态文件目录 app.mount("/static", StaticFiles(directory="static"), name="static")

在这个例子中，我们将static目录挂载到/static路径，所有对/static路径的请求都将由静态文件服务器处理。

部署与测试

部署FastAPI应用

使用Uvicorn

Uvicorn是一个ASGI服务器，适合用于开发和生产环境。

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

在这个命令中：

--host 0.0.0.0：使服务器可以从外部访问
--port 8000：指定端口为8000
--workers 4：启动4个工作进程

使用Gunicorn

对于生产环境，你可以使用Gunicorn作为进程管理器，配合Uvicorn工作器。

pip install gunicorn gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

在这个命令中：

-w 4：启动4个工作进程
-k uvicorn.workers.UvicornWorker：使用Uvicorn作为工作器
--bind 0.0.0.0:8000：绑定到所有网络接口的8000端口

使用Docker

使用Docker可以简化部署过程，并确保环境一致性。

FROM python:3.9 WORKDIR /code COPY ./requirements.txt /code/requirements.txt RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt COPY ./app /code/app CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

然后，你可以使用以下命令构建和运行Docker容器：

docker build -t myfastapi . docker run -d --name myfastapi-container -p 80:80 myfastapi

测试FastAPI应用

FastAPI提供了方便的测试工具，你可以使用TestClient来测试你的应用。

设置测试

首先，安装测试依赖：

pip install pytest httpx

然后，创建一个测试文件test_main.py：

from fastapi.testclient import TestClient from main import app client = TestClient(app) def test_read_root(): response = client.get("/") assert response.status_code == 200 assert response.json() == {"message": "Hello World"} def test_read_item(): response = client.get("/items/foo") assert response.status_code == 200 assert response.json() == { "id": "foo", "title": "Foo", "description": "There goes my hero" } def test_create_item(): response = client.post( "/items/", json={"id": "bar", "title": "Bar", "description": "The bartenders"}, ) assert response.status_code == 200 assert response.json() == { "id": "bar", "title": "Bar", "description": "The bartenders", "owner_id": 1, }

然后，运行测试：

pytest

实践案例：构建一个完整的FastAPI应用

现在，让我们构建一个完整的FastAPI应用，实现一个简单的博客系统。

项目结构

blog_app/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models.py │ ├── schemas.py │ ├── crud.py │ ├── database.py │ └── routers/ │ ├── __init__.py │ ├── posts.py │ └── users.py ├── requirements.txt └── alembic.ini

数据库模型

首先，定义数据库模型：

# app/models.py from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text, DateTime from sqlalchemy.orm import relationship from sqlalchemy.sql import func from .database import Base class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) email = Column(String, unique=True, index=True) username = Column(String, unique=True, index=True) hashed_password = Column(String) is_active = Column(Boolean, default=True) posts = relationship("Post", back_populates="author") class Post(Base): __tablename__ = "posts" id = Column(Integer, primary_key=True, index=True) title = Column(String, index=True) content = Column(Text) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) author_id = Column(Integer, ForeignKey("users.id")) author = relationship("User", back_populates="posts")

Pydantic模型

接下来，定义Pydantic模型：

# app/schemas.py from pydantic import BaseModel, EmailStr from typing import List, Optional from datetime import datetime class UserBase(BaseModel): email: EmailStr username: str class UserCreate(UserBase): password: str class UserUpdate(UserBase): password: Optional[str] = None class User(UserBase): id: int is_active: bool class Config: orm_mode = True class PostBase(BaseModel): title: str content: str class PostCreate(PostBase): pass class PostUpdate(PostBase): pass class Post(PostBase): id: int created_at: datetime updated_at: Optional[datetime] = None author_id: int author: User class Config: orm_mode = True

CRUD操作

然后，定义CRUD操作：

# app/crud.py from sqlalchemy.orm import Session from . import models, schemas from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def get_password_hash(password: str): return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str): return pwd_context.verify(plain_password, hashed_password) def get_user(db: Session, user_id: int): return db.query(models.User).filter(models.User.id == user_id).first() def get_user_by_email(db: Session, email: str): return db.query(models.User).filter(models.User.email == email).first() def get_user_by_username(db: Session, username: str): return db.query(models.User).filter(models.User.username == username).first() def create_user(db: Session, user: schemas.UserCreate): hashed_password = get_password_hash(user.password) db_user = models.User( email=user.email, username=user.username, hashed_password=hashed_password ) db.add(db_user) db.commit() db.refresh(db_user) return db_user def update_user(db: Session, user_id: int, user: schemas.UserUpdate): db_user = db.query(models.User).filter(models.User.id == user_id).first() if db_user: update_data = user.dict(exclude_unset=True) if "password" in update_data: update_data["hashed_password"] = get_password_hash(update_data.pop("password")) for field, value in update_data.items(): setattr(db_user, field, value) db.commit() db.refresh(db_user) return db_user def get_posts(db: Session, skip: int = 0, limit: int = 100): return db.query(models.Post).offset(skip).limit(limit).all() def get_post(db: Session, post_id: int): return db.query(models.Post).filter(models.Post.id == post_id).first() def create_user_post(db: Session, post: schemas.PostCreate, user_id: int): db_post = models.Post(**post.dict(), author_id=user_id) db.add(db_post) db.commit() db.refresh(db_post) return db_post def update_post(db: Session, post_id: int, post: schemas.PostUpdate): db_post = db.query(models.Post).filter(models.Post.id == post_id).first() if db_post: update_data = post.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_post, field, value) db.commit() db.refresh(db_post) return db_post def delete_post(db: Session, post_id: int): db_post = db.query(models.Post).filter(models.Post.id == post_id).first() if db_post: db.delete(db_post) db.commit() return db_post

路由

接下来，定义路由：

# app/routers/users.py from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from sqlalchemy.orm import Session from typing import List from .. import models, schemas, crud, database router = APIRouter( prefix="/users", tags=["users"] ) oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") def get_db(): db = database.SessionLocal() try: yield db finally: db.close() def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): # 这里简化了认证逻辑，实际应用中应该验证JWT令牌 user_id = 1 # 假设用户ID为1 user = crud.get_user(db, user_id=user_id) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers={"WWW-Authenticate": "Bearer"}, ) return user @router.post("/", response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): db_user = crud.get_user_by_email(db, email=user.email) if db_user: raise HTTPException(status_code=400, detail="Email already registered") db_user = crud.get_user_by_username(db, username=user.username) if db_user: raise HTTPException(status_code=400, detail="Username already registered") return crud.create_user(db=db, user=user) @router.get("/me", response_model=schemas.User) def read_users_me(current_user: models.User = Depends(get_current_user)): return current_user @router.put("/me", response_model=schemas.User) def update_user_me( user: schemas.UserUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db) ): return crud.update_user(db=db, user_id=current_user.id, user=user)

# app/routers/posts.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from typing import List from .. import models, schemas, crud, database from .users import get_current_user router = APIRouter( prefix="/posts", tags=["posts"] ) def get_db(): db = database.SessionLocal() try: yield db finally: db.close() @router.post("/", response_model=schemas.Post) def create_post( post: schemas.PostCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db) ): return crud.create_user_post(db=db, post=post, user_id=current_user.id) @router.get("/", response_model=List[schemas.Post]) def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): posts = crud.get_posts(db, skip=skip, limit=limit) return posts @router.get("/117629", response_model=schemas.Post) def read_post(post_id: int, db: Session = Depends(get_db)): db_post = crud.get_post(db, post_id=post_id) if db_post is None: raise HTTPException(status_code=404, detail="Post not found") return db_post @router.put("/117629", response_model=schemas.Post) def update_post( post_id: int, post: schemas.PostUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db) ): db_post = crud.get_post(db, post_id=post_id) if db_post is None: raise HTTPException(status_code=404, detail="Post not found") if db_post.author_id != current_user.id: raise HTTPException(status_code=403, detail="Not authorized to update this post") return crud.update_post(db=db, post_id=post_id, post=post) @router.delete("/117629") def delete_post( post_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db) ): db_post = crud.get_post(db, post_id=post_id) if db_post is None: raise HTTPException(status_code=404, detail="Post not found") if db_post.author_id != current_user.id: raise HTTPException(status_code=403, detail="Not authorized to delete this post") crud.delete_post(db=db, post_id=post_id) return {"detail": "Post deleted successfully"}

主应用

最后，定义主应用：

# app/main.py from fastapi import FastAPI from .routers import users, posts from . import database # 创建数据库表 database.Base.metadata.create_all(bind=database.engine) app = FastAPI(title="Blog API") # 包含路由 app.include_router(users.router) app.include_router(posts.router) @app.get("/") def read_root(): return {"message": "Welcome to Blog API"}