FastAPI和Uvicorn协同工作原理详解打造高性能异步Web应用从安装配置到部署上线一步步教你构建可扩展的API服务

1. 引言

FastAPI是一个现代、高性能的Python Web框架，专为构建API而设计。它基于Python的类型提示和异步编程模型，提供了快速开发、高性能和易于维护的特性。而Uvicorn是一个ASGI（Asynchronous Server Gateway Interface）服务器，用于运行异步Python Web应用。FastAPI和Uvicorn的结合为开发者提供了一个强大而高效的工具组合，用于构建可扩展的API服务。

2. FastAPI和Uvicorn基础概念

2.1 FastAPI核心特性

FastAPI具有以下核心特性：

高性能：基于Starlette（ASGI框架）和Pydantic（数据验证库），性能接近Node.js和Go。
异步支持：完全支持异步编程（async和await），能够高效处理并发请求。
类型提示与数据验证：利用Python的类型提示（Type Hints）和Pydantic模型，自动验证请求和响应数据。
自动生成API文档：自动生成交互式API文档，支持Swagger UI和ReDoc。
依赖注入系统：提供了强大的依赖注入机制，方便管理共享逻辑。
标准兼容：完全兼容OpenAPI（以前称为Swagger）和JSON Schema标准。

2.2 Uvicorn简介

Uvicorn是一个轻量级的ASGI服务器，具有以下特点：

基于uvloop和httptools，提供极高的性能
支持HTTP/1.1和WebSockets
支持异步请求处理
可以作为独立服务器运行，也可以与其他进程管理器（如Gunicorn）结合使用

2.3 ASGI与WSGI的区别

ASGI（Asynchronous Server Gateway Interface）是WSGI（Web Server Gateway Interface）的异步演进版本。主要区别在于：

WSGI是同步的，每个请求在一个独立的线程中处理
ASGI支持异步处理，可以在单个线程中处理多个并发请求
ASGI支持WebSocket等协议，而WSGI仅支持HTTP

3. 安装与环境配置

3.1 创建虚拟环境

首先，我们需要创建一个独立的Python虚拟环境：

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

3.2 安装FastAPI和Uvicorn

在激活的虚拟环境中，安装FastAPI和Uvicorn：

# 安装FastAPI和Uvicorn pip install fastapi "uvicorn[standard]" # 安装其他有用的依赖 pip install pydantic python-multipart

[standard]选项会安装Uvicorn及其推荐的依赖项，包括websockets（用于WebSocket支持）和httptools（用于HTTP解析优化）。

3.3 项目结构

创建一个基本的项目结构：

fastapi_project/ ├── app/ │ ├── __init__.py │ ├── main.py # 主应用文件 │ ├── routers/ # 路由模块 │ ├── models/ # 数据模型 │ ├── schemas/ # Pydantic模式 │ └── dependencies.py # 依赖项 ├── tests/ # 测试文件 ├── requirements.txt # 依赖项列表 └── .env # 环境变量

4. 构建基础FastAPI应用

4.1 创建简单的FastAPI应用

让我们创建一个简单的FastAPI应用：

# app/main.py from fastapi import FastAPI app = FastAPI(title="我的第一个FastAPI应用", version="1.0.0") @app.get("/") async def read_root(): return {"message": "Hello, World!"} @app.get("/items/{item_id}") async def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q}

4.2 运行FastAPI应用

使用Uvicorn运行FastAPI应用：

# 基本运行命令 uvicorn app.main:app --reload # 指定主机和端口 uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload # 在生产环境中运行（不使用reload） uvicorn app.main:app --host 0.0.0.0 --port 8000

参数说明：

app.main:app：指定应用的位置，格式为模块:应用对象
--reload：启用自动重载，代码更改后服务器会自动重启（开发环境推荐）
--host 0.0.0.0：使服务器可以从外部访问
--port 8000：指定端口号

4.3 访问自动生成的API文档

启动应用后，你可以访问以下URL查看自动生成的API文档：

Swagger UI: http://127.0.0.1:8000/docs
ReDoc: http://127.0.0.1:8000/redoc

5. 异步编程原理与实践

5.1 异步编程基础

FastAPI充分利用了Python的异步编程能力。在异步编程中，当一个任务等待I/O操作（如数据库查询、HTTP请求）完成时，它可以释放控制权，让其他任务运行，从而提高并发性能。

5.2 异步路由处理

在FastAPI中，你可以使用async def定义异步路由处理函数：

import asyncio from fastapi import FastAPI app = FastAPI() @app.get("/sync") def sync_route(): # 同步路由处理函数 return {"message": "这是一个同步路由"} @app.get("/async") async def async_route(): # 异步路由处理函数 await asyncio.sleep(1) # 模拟I/O操作 return {"message": "这是一个异步路由"}

5.3 异步数据库操作

以异步数据库操作为例，展示如何使用异步编程提高性能：

import asyncio from databases import Database from fastapi import FastAPI app = FastAPI() # 创建数据库连接 database = Database("postgresql://user:password@localhost/dbname") @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() @app.get("/users/{user_id}") async def read_user(user_id: int): query = "SELECT * FROM users WHERE id = :user_id" result = await database.fetch_one(query, {"user_id": user_id}) return {"user": result}

6. 路由、请求处理与响应

6.1 路由定义

FastAPI使用装饰器定义路由，支持所有HTTP方法：

from fastapi import FastAPI, HTTPException app = FastAPI() @app.get("/items/{item_id}") async def read_item(item_id: int): return {"item_id": item_id} @app.post("/items/") async def create_item(item: dict): return {"item": item} @app.put("/items/{item_id}") async def update_item(item_id: int, item: dict): return {"item_id": item_id, "updated_item": item} @app.delete("/items/{item_id}") async def delete_item(item_id: int): return {"item_id": item_id, "status": "deleted"}

6.2 路径参数和查询参数

FastAPI支持路径参数和查询参数，并自动进行类型验证：

from fastapi import FastAPI from typing import Optional app = FastAPI() @app.get("/users/{user_id}") async def read_user(user_id: int, q: Optional[str] = None, short: bool = False): user = {"user_id": user_id} if q: user.update({"q": q}) if not short: user.update({"description": "这是一个用户描述"}) return user

6.3 请求体处理

使用Pydantic模型处理请求体：

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

6.4 响应模型

使用响应模型来规范输出：

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Item(BaseModel): name: str description: Optional[str] = None price: float tax: float = 10.5 class ResponseItem(BaseModel): name: str 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, "price_with_tax": price_with_tax}

7. 数据验证与序列化

7.1 Pydantic模型

Pydantic是FastAPI中用于数据验证和序列化的核心库：

from pydantic import BaseModel, EmailStr, Field from typing import List, Optional from datetime import datetime class User(BaseModel): id: Optional[int] = None name: str = Field(..., min_length=2, max_length=50) email: EmailStr age: int = Field(..., ge=18, le=120) is_active: bool = True created_at: datetime = None friends: List[int] = [] class Config: schema_extra = { "example": { "name": "John Doe", "email": "john@example.com", "age": 30, "is_active": True, "friends": [1, 2, 3] } }

7.2 数据验证

FastAPI会自动验证输入数据：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel, constr app = FastAPI() class User(BaseModel): username: constr(min_length=3, max_length=20) password: constr(min_length=8) @app.post("/users/") async def create_user(user: User): # 如果验证失败，FastAPI会自动返回422错误 # 这里只有验证通过才会执行 return {"username": user.username}

7.3 数据转换

Pydantic模型可以轻松地将数据转换为不同格式：

from pydantic import BaseModel from datetime import datetime class User(BaseModel): id: int name: str signup_ts: datetime # 从字典创建模型实例 user_data = {"id": 123, "name": "John", "signup_ts": "2023-01-01 12:00"} user = User(**user_data) # 转换为字典 user_dict = user.dict() # 转换为JSON user_json = user.json()

8. 依赖注入系统

8.1 基本依赖注入

FastAPI的依赖注入系统允许你定义组件的依赖关系：

from fastapi import FastAPI, Depends app = FastAPI() async def common_parameters(q: str = 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

8.2 类作为依赖

你也可以使用类作为依赖：

from fastapi import FastAPI, Depends from typing import Optional app = FastAPI() class CommonQueryParams: def __init__(self, q: Optional[str] = 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

8.3 带状态的依赖

使用yield创建有状态的依赖，例如数据库连接：

from fastapi import FastAPI, Depends from databases import Database app = FastAPI() async def get_db(): database = Database("postgresql://user:password@localhost/dbname") await database.connect() try: yield database finally: await database.disconnect() @app.get("/users/{user_id}") async def read_user(user_id: int, db: Database = Depends(get_db)): query = "SELECT * FROM users WHERE id = :user_id" result = await db.fetch_one(query, {"user_id": user_id}) return {"user": result}

9. 中间件与异常处理

9.1 中间件

FastAPI支持中间件，可以在请求处理前后执行代码：

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

9.2 CORS中间件

处理跨域资源共享（CORS）：

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )

9.3 异常处理

自定义异常处理：

from fastapi import FastAPI, Request, HTTPException 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}, ) @app.get("/items/{item_id}") async def read_item(item_id: int): if item_id < 0: raise HTTPException(status_code=400, detail="Item ID must be positive") return {"item_id": item_id}

10. 测试FastAPI应用

10.1 使用TestClient进行测试

FastAPI提供了TestClient用于测试：

from fastapi import FastAPI from fastapi.testclient import TestClient app = FastAPI() @app.get("/") async def read_root(): return {"message": "Hello World"} client = TestClient(app) def test_read_root(): response = client.get("/") assert response.status_code == 200 assert response.json() == {"message": "Hello World"}

10.2 测试异步路由

测试异步路由：

import pytest from httpx import AsyncClient from app.main import app @pytest.mark.asyncio async def test_async_route(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/async") assert response.status_code == 200 assert response.json() == {"message": "这是一个异步路由"}

10.3 测试数据库操作

测试涉及数据库的路由：

import pytest from httpx import AsyncClient from app.main import app from app.database import get_database from app.test_database import override_get_database app.dependency_overrides[get_database] = override_get_database @pytest.mark.asyncio async def test_create_user(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.post( "/users/", json={"name": "John", "email": "john@example.com", "age": 30} ) assert response.status_code == 200 assert response.json()["name"] == "John"

11. 性能优化与扩展

11.1 使用Gunicorn和Uvicorn Workers

在生产环境中，可以使用Gunicorn管理多个Uvicorn工作进程：

gunicorn app.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端口

11.2 使用Redis缓存

使用Redis缓存提高性能：

from fastapi import FastAPI, Depends import redis import json app = FastAPI() redis_client = redis.Redis(host='localhost', port=6379, db=0) def get_cache(key: str): cached_data = redis_client.get(key) if cached_data: return json.loads(cached_data) return None def set_cache(key: str, value: dict, expire: int = 3600): redis_client.setex(key, expire, json.dumps(value)) @app.get("/users/{user_id}") async def read_user(user_id: int): cache_key = f"user:{user_id}" user_data = get_cache(cache_key) if user_data: return {"user": user_data, "from_cache": True} # 模拟数据库查询 user_data = {"id": user_id, "name": f"User {user_id}"} set_cache(cache_key, user_data) return {"user": user_data, "from_cache": False}

11.3 异步任务处理

使用Celery处理耗时任务：

# tasks.py from celery import Celery celery_app = Celery( "worker", broker="redis://localhost:6379/0", backend="redis://localhost:6379/0" ) @celery_app.task def process_data(data: dict): # 模拟耗时操作 import time time.sleep(5) return {"processed_data": data, "status": "completed"} # main.py from fastapi import FastAPI, BackgroundTasks from tasks import process_data app = FastAPI() @app.post("/process/") async def process_endpoint(data: dict, background_tasks: BackgroundTasks): background_tasks.add_task(process_data.delay, data) return {"message": "Processing started"}

12. 部署上线

12.1 使用Docker容器化

创建Dockerfile：

# 使用官方Python运行时作为父镜像 FROM python:3.9-slim # 设置工作目录 WORKDIR /app # 复制依赖文件 COPY requirements.txt . # 安装依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 暴露端口 EXPOSE 8000 # 运行应用 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

构建和运行Docker容器：

# 构建镜像 docker build -t fastapi-app . # 运行容器 docker run -d --name fastapi-container -p 8000:8000 fastapi-app

12.2 使用Docker Compose

使用Docker Compose管理多容器应用：

# docker-compose.yml version: "3.8" services: api: build: . ports: - "8000:8000" depends_on: - redis - db environment: - DATABASE_URL=postgresql://fastapi:fastapi@db:5432/fastapi_db - REDIS_URL=redis://redis:6379 redis: image: redis:alpine ports: - "6379:6379" db: image: postgres:13 environment: - POSTGRES_USER=fastapi - POSTGRES_PASSWORD=fastapi - POSTGRES_DB=fastapi_db volumes: - postgres_data:/var/lib/postgresql/data ports: - "5432:5432" volumes: postgres_data:

启动服务：

docker-compose up -d

12.3 使用Nginx作为反向代理

配置Nginx作为反向代理：

server { listen 80; server_name yourdomain.com; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } location /docs { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } }

13. 实战案例：构建一个完整的API服务

让我们构建一个完整的博客API服务，包括用户认证、文章管理和评论功能。

13.1 项目结构

blog_api/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py # 配置文件 │ ├── database.py # 数据库连接 │ ├── models/ │ │ ├── __init__.py │ │ ├── user.py │ │ ├── post.py │ │ └── comment.py │ ├── schemas/ │ │ ├── __init__.py │ │ ├── user.py │ │ ├── post.py │ │ └── comment.py │ ├── routers/ │ │ ├── __init__.py │ │ ├── users.py │ │ ├── posts.py │ │ └── comments.py │ ├── auth.py # 认证相关 │ └── dependencies.py # 依赖项 ├── tests/ ├── alembic/ # 数据库迁移 ├── requirements.txt ├── Dockerfile └── docker-compose.yml

13.2 数据库模型

定义用户、文章和评论模型：

# app/models/user.py from sqlalchemy import Column, Integer, String, Boolean, DateTime from sqlalchemy.sql import func from app.database import Base class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String, unique=True, index=True) email = Column(String, unique=True, index=True) hashed_password = Column(String) is_active = Column(Boolean, default=True) created_at = Column(DateTime(timezone=True), server_default=func.now())

# app/models/post.py from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey from sqlalchemy.sql import func from sqlalchemy.orm import relationship from app.database import Base class Post(Base): __tablename__ = "posts" id = Column(Integer, primary_key=True, index=True) title = Column(String, index=True) content = Column(Text) owner_id = Column(Integer, ForeignKey("users.id")) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) owner = relationship("User", back_populates="posts") comments = relationship("Comment", back_populates="post")

# app/models/comment.py from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey from sqlalchemy.sql import func from sqlalchemy.orm import relationship from app.database import Base class Comment(Base): __tablename__ = "comments" id = Column(Integer, primary_key=True, index=True) content = Column(Text) post_id = Column(Integer, ForeignKey("posts.id")) owner_id = Column(Integer, ForeignKey("users.id")) created_at = Column(DateTime(timezone=True), server_default=func.now()) owner = relationship("User", back_populates="comments") post = relationship("Post", back_populates="comments")

13.3 Pydantic模式

定义请求和响应模式：

# app/schemas/user.py from pydantic import BaseModel, EmailStr from typing import Optional from datetime import datetime class UserBase(BaseModel): username: str email: EmailStr class UserCreate(UserBase): password: str class UserUpdate(UserBase): password: Optional[str] = None class User(UserBase): id: int is_active: bool created_at: datetime class Config: orm_mode = True

# app/schemas/post.py from pydantic import BaseModel from typing import Optional, List from datetime import datetime class PostBase(BaseModel): title: str content: str class PostCreate(PostBase): pass class PostUpdate(PostBase): title: Optional[str] = None content: Optional[str] = None class Post(PostBase): id: int owner_id: int created_at: datetime updated_at: Optional[datetime] = None class Config: orm_mode = True

# app/schemas/comment.py from pydantic import BaseModel from typing import Optional from datetime import datetime class CommentBase(BaseModel): content: str class CommentCreate(CommentBase): post_id: int class CommentUpdate(CommentBase): content: Optional[str] = None class Comment(CommentBase): id: int post_id: int owner_id: int created_at: datetime class Config: orm_mode = True

13.4 认证系统

实现JWT认证：

# app/auth.py from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from app.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt async def get_current_user(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception except JWTError: raise credentials_exception # 这里应该从数据库获取用户 user = get_user(username=username) if user is None: raise credentials_exception return user

13.5 路由定义

定义用户、文章和评论的路由：

# app/routers/users.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.database import get_db from app.models import User as UserModel from app.schemas import User, UserCreate, UserUpdate from app.auth import get_current_user, get_password_hash router = APIRouter( prefix="/users", tags=["users"] ) @router.post("/", response_model=User) async def create_user(user: UserCreate, db: Session = Depends(get_db)): db_user = db.query(UserModel).filter(UserModel.email == user.email).first() if db_user: raise HTTPException( status_code=400, detail="Email already registered" ) hashed_password = get_password_hash(user.password) db_user = UserModel( username=user.username, email=user.email, hashed_password=hashed_password ) db.add(db_user) db.commit() db.refresh(db_user) return db_user @router.get("/me", response_model=User) async def read_users_me(current_user: User = Depends(get_current_user)): return current_user @router.put("/me", response_model=User) async def update_user_me( user_update: UserUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): db_user = db.query(UserModel).filter(UserModel.id == current_user.id).first() update_data = user_update.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

# app/routers/posts.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from typing import List from app.database import get_db from app.models import Post as PostModel, User as UserModel from app.schemas import Post, PostCreate, PostUpdate from app.auth import get_current_user router = APIRouter( prefix="/posts", tags=["posts"] ) @router.post("/", response_model=Post) async def create_post( post: PostCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): db_post = PostModel(**post.dict(), owner_id=current_user.id) db.add(db_post) db.commit() db.refresh(db_post) return db_post @router.get("/", response_model=List[Post]) async def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): posts = db.query(PostModel).offset(skip).limit(limit).all() return posts @router.get("/117590", response_model=Post) async def read_post(post_id: int, db: Session = Depends(get_db)): db_post = db.query(PostModel).filter(PostModel.id == post_id).first() if db_post is None: raise HTTPException(status_code=404, detail="Post not found") return db_post @router.put("/117590", response_model=Post) async def update_post( post_id: int, post_update: PostUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): db_post = db.query(PostModel).filter(PostModel.id == post_id).first() if db_post is None: raise HTTPException(status_code=404, detail="Post not found") if db_post.owner_id != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to update this post" ) update_data = post_update.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 @router.delete("/117590") async def delete_post( post_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): db_post = db.query(PostModel).filter(PostModel.id == post_id).first() if db_post is None: raise HTTPException(status_code=404, detail="Post not found") if db_post.owner_id != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to delete this post" ) db.delete(db_post) db.commit() return {"detail": "Post deleted"}

# app/routers/comments.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from typing import List from app.database import get_db from app.models import Comment as CommentModel, User as UserModel, Post as PostModel from app.schemas import Comment, CommentCreate, CommentUpdate from app.auth import get_current_user router = APIRouter( prefix="/comments", tags=["comments"] ) @router.post("/", response_model=Comment) async def create_comment( comment: CommentCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): # 检查帖子是否存在 db_post = db.query(PostModel).filter(PostModel.id == comment.post_id).first() if db_post is None: raise HTTPException(status_code=404, detail="Post not found") db_comment = CommentModel( content=comment.content, post_id=comment.post_id, owner_id=current_user.id ) db.add(db_comment) db.commit() db.refresh(db_comment) return db_comment @router.get("/post/117590", response_model=List[Comment]) async def read_comments_by_post(post_id: int, db: Session = Depends(get_db)): # 检查帖子是否存在 db_post = db.query(PostModel).filter(PostModel.id == post_id).first() if db_post is None: raise HTTPException(status_code=404, detail="Post not found") comments = db.query(CommentModel).filter(CommentModel.post_id == post_id).all() return comments @router.put("/{comment_id}", response_model=Comment) async def update_comment( comment_id: int, comment_update: CommentUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): db_comment = db.query(CommentModel).filter(CommentModel.id == comment_id).first() if db_comment is None: raise HTTPException(status_code=404, detail="Comment not found") if db_comment.owner_id != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to update this comment" ) update_data = comment_update.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_comment, field, value) db.commit() db.refresh(db_comment) return db_comment @router.delete("/{comment_id}") async def delete_comment( comment_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): db_comment = db.query(CommentModel).filter(CommentModel.id == comment_id).first() if db_comment is None: raise HTTPException(status_code=404, detail="Comment not found") if db_comment.owner_id != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to delete this comment" ) db.delete(db_comment) db.commit() return {"detail": "Comment deleted"}

13.6 主应用文件

将所有路由组合到主应用中：

# app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.routers import users, posts, comments from app.database import engine from app.models import Base # 创建数据库表 Base.metadata.create_all(bind=engine) app = FastAPI( title="博客API", description="一个使用FastAPI构建的简单博客API", version="1.0.0" ) # 配置CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 注册路由 app.include_router(users.router) app.include_router(posts.router) app.include_router(comments.router) @app.get("/") async def root(): return {"message": "欢迎来到博客API"} @app.get("/health") async def health_check(): return {"status": "ok"}

13.7 运行和测试

使用Uvicorn运行应用：

uvicorn app.main:app --reload

然后你可以访问http://127.0.0.1:8000/docs查看自动生成的API文档并进行测试。

13.8 部署到生产环境

使用Docker Compose部署到生产环境：

# docker-compose.yml version: "3.8" services: api: build: . ports: - "8000:8000" depends_on: - db environment: - DATABASE_URL=postgresql://fastapi:fastapi@db:5432/fastapi_db command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 db: image: postgres:13 environment: - POSTGRES_USER=fastapi - POSTGRES_PASSWORD=fastapi - POSTGRES_DB=fastapi_db volumes: - postgres_data:/var/lib/postgresql/data ports: - "5432:5432" nginx: image: nginx:alpine ports: - "80:80" volumes: - ./nginx.conf:/etc/nginx/nginx.conf depends_on: - api volumes: postgres_data:

Nginx配置文件：

# nginx.conf events { worker_connections 1024; } http { upstream api { server api:8000; } server { listen 80; location / { proxy_pass http://api; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } } }

启动服务：