Skip to main content

Python 开发 - Web开发

Python Web 开发：FastAPI 实战指南

引言

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API。它基于标准的 Python 类型提示，并提供了许多开箱即用的功能，如自动交互式文档、数据验证和依赖注入。本文将以一个实际的 AI 招聘项目为例，深入探讨 FastAPI 的核心概念和最佳实践。

项目结构

一个组织良好的项目结构是高效开发的基础。下面是我们示例项目的目录结构：

.
├── api
│   └── v1
│       ├── auth.py
│       ├── recuitment.py
│       └── ...
├── celery_server
│   ├── celery_app.py
│   └── tasks.py
├── schema
│   ├── aitask
│   ├── recuitment
│   └── ...
├── utils
│   ├── connections
│   │   ├── async_milvus.py
│   │   └── ...
│   └── tools.py
├── config.py
├── dependencies.py
├── main.py
├── requirements.txt
└── test
  • main.py: 应用入口，负责初始化 FastAPI 应用和中间件。
  • config.py: 存放应用的配置信息，如数据库连接字符串、API 密钥等。
  • dependencies.py: 定义了项目范围内的依赖注入项。
  • requirements.txt: 列出了项目的所有 Python 依赖。
  • api/v1/: 存放不同版本的 API 路由。这种组织方式便于 API 的版本管理。
  • schema/: 包含了所有的 Pydantic 模型，用于请求和响应的数据验证与序列化。
  • utils/: 包含工具函数和数据库连接模块。
  • celery_server/: 存放 Celery 相关的配置和异步任务。
  • test/: 包含集成测试和单元测试。

核心概念

应用启动与生命周期

FastAPI 提供了生命周期事件，允许我们在应用启动和关闭时执行代码。在我们的项目中，我们使用 @asynccontextmanager 来管理应用的生命周期，这是一种更现代、更 Pythonic 的方式。

// main.py

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用启动时执行
    print("Application startup")
    app.state.db_connection = await connect_to_db()
    yield
    # 应用关闭时执行
    await app.state.db_connection.close()
    print("Application shutdown")

app = FastAPI(lifespan=lifespan)

在我们的实际项目中，lifespan 函数负责初始化和关闭数据库连接池（MySQL, Redis, Milvus）以及其他服务（如 MinIO 和 OpenAI）。

依赖注入

依赖注入是 FastAPI 的核心功能之一。它允许我们将依赖项（如数据库会话、认证用户等）声明为函数参数，FastAPI 会在处理请求时自动创建和管理这些依赖。

// dependencies.py

from fastapi import Depends, HTTPException, status
from .utils.connections import get_db_session

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # ... 从 token 中获取用户 ...
    return user

def get_database_session():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

在路由函数中，我们可以像这样使用依赖项：

// api/v1/recuitment.py

from fastapi import APIRouter, Depends
from ..dependencies import get_current_user

router = APIRouter()

@router.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

路由与 API 版本管理

将路由分散到不同的模块中，可以使代码更加清晰和易于维护。我们的项目通过 APIRouter 来实现这一点，并将所有路由集中在 api/v1/ 目录下。

// main.py

from api.v1 import all_routers

app = FastAPI()
common_prefix = "/api/v1" 

for router_instance in all_routers:
    app.include_router(router_instance, prefix=common_prefix)

这种方式不仅使得代码结构清晰，也为未来的 API 版本升级（如 /api/v2）提供了便利。

数据模型与验证

FastAPI 使用 Pydantic 模型进行数据验证、序列化和文档生成。通过在 schema/ 目录中为不同的业务逻辑定义模型，我们可以确保 API 的输入和输出都符合预期的格式。

// schema/recuitment/PositionBaseData.py

from pydantic import BaseModel
from typing import Optional

class PositionBaseData(BaseModel):
    title: str
    description: Optional[str] = None
    salary: float

在 API 操作中，我们可以直接使用这些模型作为类型提示：

// api/v1/recuitment.py

from fastapi import APIRouter
from ...schema.recuitment import PositionBaseData

router = APIRouter()

@router.post("/positions/")
async def create_position(position: PositionBaseData):
    # ... 创建职位的逻辑 ...
    return position

配置文件管理

使用独立的配置文件来管理应用的设置是一种最佳实践。在我们的项目中，config.py 使用了 pydantic-settings 库，它可以从环境变量或 .env 文件中加载配置。

// config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "AI Recruitment API"
    database_url: str
    openai_api_key: str

    class Config:
        env_file = ".env"

settings = Settings()

异步任务

对于耗时的操作，如发送邮件、处理文件等，我们应该使用异步任务来避免阻塞主线程。Celery 是一个流行的任务队列，可以与 FastAPI 很好地集成。

在 celery_server/ 目录下，我们定义了 Celery 应用实例和任务：

// celery_server/celery_app.py

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379/0", backend="redis://localhost:6379/0")

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Shanghai",
    enable_utc=True,
)

// celery_server/tasks.py

from .celery_app import app

@app.task
def process_resume(resume_id: int):
    # ... 处理简历的耗时逻辑 ...
    return {"status": "success", "resume_id": resume_id}

数据库集成

我们的项目集成了多种数据库和存储服务，并通过 utils/connections 模块进行统一管理。这种模块化的设计使得数据库的切换和扩展变得更加容易。

  • MySQL: 使用 aiomysql 提供异步连接池。
  • Redis: 用于缓存和 Celery 的消息代理。
  • Milvus: 一个开源的向量数据库，用于 AI 相关的相似性搜索。
  • MinIO: 一个兼容 S3 的对象存储服务，用于存储文件（如简历）。

通过在应用的生命周期中管理这些连接，我们确保了资源的有效利用和优雅关闭。

测试

测试是保证软件质量的关键。我们的项目包含了 test/ 和 unit-test/ 两个目录，分别用于集成测试和单元测试。FastAPI 提供了 TestClient，可以方便地对 API 进行测试。

// test/test_main.py

from fastapi.testclient import TestClient
from ..main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

总结

FastAPI 凭借其卓越的性能、易用性和强大的功能，已成为现代 Python Web 开发的热门选择。通过本文对一个实际项目的剖析，我们深入了解了 FastAPI 在项目结构、依赖注入、数据验证、异步任务和数据库集成等方面的最佳实践。希望这篇指南能为你的 FastAPI 开发之旅提供有益的参考。

No comments to display

Back to top