FastAPI 启动开发框架

这是一个基于 FastAPI 构建的后端 API 基础框架，遵循最佳实践，为快速开发高性能的 Web API 提供坚实基础。

✨ 功能特性

模块化结构：按功能域组织代码，提高可维护性
开发者管理：开发者注册与信息管理
API 密钥管理：安全的 API 密钥生成、分发与生命周期管理
JWT 认证：基于 JWT 的请求认证机制
异步支持：全面利用 FastAPI 的异步功能，提高性能
ORM 集成：完全集成 Tortoise ORM，提供强大的异步数据库交互
环境配置：灵活的多环境配置系统
日志系统：集成 Loguru，提供优雅的日志记录
CLI 工具：完整的命令行工具集，用于管理开发者、API密钥和数据库迁移

🚀 快速开始

前提条件

Python 3.8+
MySQL 5.7+
Poetry (Python 依赖管理工具)
Docker 和 Docker Compose (可选，用于容器化部署)

安装依赖

# 安装项目依赖
poetry install

# 方法 1：使用 poetry run 运行命令（推荐，适用于所有 Poetry 版本）
poetry run python -m src

# 方法 2：手动激活虚拟环境（如果需要在虚拟环境中执行多个命令）
source $(poetry env info --path)/bin/activate

配置环境变量

复制 .env.example 文件并重命名为 .env，然后根据您的环境进行配置：

# 应用程序设置
APP_NAME=fastapi-base-framework
HOST=0.0.0.0
PORT=8000
WORKERS_COUNT=1
RELOAD=True
ENVIRONMENT=development
DEBUG=True
LOG_LEVEL=INFO

# 数据库设置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=
DB_BASE=fastapi_base
DB_ECHO=False

# JWT设置
SECRET_KEY=replace-with-secure-secret-key
JWT_SECRET=replace-with-secure-jwt-secret
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=1440

初始化数据库

# 创建数据库
mysql -u root -e "CREATE DATABASE IF NOT EXISTS fastapi_base CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"

# 使用初始化脚本创建数据库结构
python scripts/init_db.py

# 或者使用 CLI 工具分步操作

# 初始化数据库迁移环境
python -m src.cli.main db init

# 生成迁移文件
python -m src.cli.main db migrate "初始迁移"

# 应用迁移
python -m src.cli.main db upgrade

运行开发服务器

poetry run python -m src

服务启动后，您可以访问以下地址：

API 文档: http://localhost:8000/api/docs
Redoc 文档: http://localhost:8000/api/redoc

🐳 使用 Docker 运行

开发模式

docker-compose -f docker-compose.yml -f deploy/docker-compose.dev.yml up --build

生产模式

docker-compose up -d --build

🛠️ CLI 工具使用指南

本项目提供了一个强大的命令行工具，用于管理开发者、API密钥和数据库迁移等操作。CLI 工具基于 Typer 构建，提供了友好的命令行界面和详细的帮助信息。

基本使用

# 显示主帮助信息和可用命令
python -m src.cli

# 或使用 Poetry
poetry run python -m src.cli

开发者管理命令

# 查看开发者命令帮助
python -m src.cli developer --help

# 列出所有开发者
python -m src.cli developer list

# 添加新开发者
python -m src.cli developer add --name "开发者名称" --email "developer@example.com"
# 简写形式
python -m src.cli developer add -n "开发者名称" -e "developer@example.com"

# 获取开发者详情
python -m src.cli developer get 1

# 更新开发者信息
python -m src.cli developer update 1 --name "新名称" --email "new-email@example.com"

# 删除开发者
python -m src.cli developer delete 1
# 跳过确认
python -m src.cli developer delete 1 -y

API 密钥管理命令

# 查看 API 密钥命令帮助
python -m src.cli api-key --help

# 列出开发者的所有 API 密钥
python -m src.cli api-key list 1

# 为开发者创建新的 API 密钥
python -m src.cli api-key create 1

# 撤销 API 密钥
python -m src.cli api-key revoke 123
# 跳过确认
python -m src.cli api-key revoke 123 -y

数据库管理命令

# 查看数据库命令帮助
python -m src.cli db --help

🔧 Aerich 数据迁移使用指南

本项目使用 Aerich 进行数据库迁移管理，下面是详细的使用步骤。

初始化数据库

在第一次运行项目时，您需要初始化数据库结构。我们提供了两种方式：

方式一：使用初始化脚本

# 创建数据库
mysql -u root -p -e "CREATE DATABASE IF NOT EXISTS fastapi_base CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"

# 运行初始化脚本
python scripts/init_db.py

初始化脚本会自动完成以下操作：

测试数据库连接
初始化 Aerich 环境
初始化数据库结构
创建迁移记录

方式二：使用 CLI 命令逐步操作

# 初始化 Aerich 配置
python -m src.cli.main db init

# 生成初始化迁移文件
python -m src.cli.main db migrate "初始化数据库"

# 应用迁移
python -m src.cli.main db upgrade

创建新的数据库迁移

当您对数据库模型进行修改后，需要创建新的迁移文件：

# 生成新的迁移文件
python -m src.cli.main db migrate "描述您的更改"

# 应用新的迁移
python -m src.cli.main db upgrade

管理数据库迁移

查看迁移历史

python -m src.cli.main db history

回滚迁移

# 回滚到上一个版本
python -m src.cli.main db downgrade

# 或者回滚到指定版本
python -m src.cli.main db downgrade "版本名称"

查看数据库模型结构

python -m src.cli.main db show-models

Aerich 配置文件

项目根目录下的 aerich.ini 文件定义了 Aerich 的配置：

[aerich]
tortoise_orm = src.db.config.TORTOISE_CONFIG
location = ./migrations
src_folder = ./

tortoise_orm: 指向 Tortoise ORM 配置对象的路径
location: 迁移文件存储位置
src_folder: 源代码文件夹路径

注意事项

保持模型引用一致性: 在数据库模型中的外键关系中，使用一致的引用方式，如 "models.ModelName"。
安全变更迁移: 对于复杂的数据库变更，考虑手动编辑迁移文件以确保安全迁移。
测试环境迁移: 在应用到生产环境前，在测试环境中测试所有迁移。
迁移回滚注意事项: 回滚数据库迁移可能会导致数据丢失，请谨慎操作，并在运行前备份数据库。

📚 API 文档

服务启动后，您可以访问如下 API 文档：

Swagger UI: http://localhost:8000/api/docs
ReDoc: http://localhost:8000/api/redoc

主要接口

认证

POST /api/auth/token - 获取访问令牌

健康检查

GET /api/health - 获取系统状态信息
GET /api/health/protected - 获取需要认证的系统状态信息（需要 JWT 认证）

🛠️ 开发指南

项目结构

src/
├── api/               # API 相关组件
│   ├── auth/           # 认证模块
│   │   ├── constants.py  # 常量定义
│   │   ├── endpoints.py  # 路由定义
│   │   ├── schemas.py    # 请求/响应模型
│   │   └── service.py    # 业务逻辑
│   ├── health/         # 健康检查模块
│   │   ├── constants.py  # 常量定义
│   │   ├── endpoints.py  # 路由定义
│   │   └── schemas.py    # 请求/响应模型
│   ├── dependencies.py # 共享依赖定义
│   └── router.py      # API 路由注册
├── cli/               # 命令行工具
│   ├── commands/       # 各类命令定义
│   ├── db.py           # 数据库管理命令
│   └── main.py         # CLI 主入口
├── core/              # 核心组件
│   ├── base_models.py  # 基础模型定义
│   ├── config.py       # 应用配置
│   ├── exceptions.py   # 异常定义
│   ├── log.py          # 日志配置
│   └── security.py     # 安全相关功能
├── db/                # 数据库组件
│   ├── migrations/     # 数据库迁移文件
│   ├── models/         # 数据库模型
│   │   ├── api_key.py    # API密钥模型
│   │   ├── base.py       # 基础模型
│   │   └── developer.py  # 开发者模型
│   └── config.py       # 数据库配置
├── app.py             # FastAPI 应用实例
└── __main__.py        # 应用入口

scripts/                # 实用脚本
└── init_db.py        # 数据库初始化脚本

按功能域组织代码

本框架遵循"按功能域组织代码"的原则，将相关功能代码组织在一起，而不是按文件类型分散。在 src/api 目录下，每个功能模块包含以下文件：

__init__.py: 模块定义和公共API导出
endpoints.py: 路由处理函数定义
schemas.py: 请求和响应数据模型
service.py: 模块特定的业务逻辑
constants.py: 模块特定的常量和错误代码
exceptions.py: 模块特定的异常类
utils.py: 非业务逻辑的工具函数

例如，认证模块的组织结构：

src/api/auth/
├── __init__.py      # 导出 router, TokenResponse 等公共API
├── constants.py     # 定义TOKEN_TYPE等常量
├── endpoints.py     # 定义路由处理函数
├── exceptions.py    # 定义InvalidApiKeyError等异常
├── schemas.py       # 定义TokenRequest等数据模型
└── service.py       # 实现authenticate_api_key等业务逻辑

数据库模型集中存放在 src/db/models 目录下，使用 Tortoise ORM 定义。每个模型类都在单独的文件中定义，并包含必要的Pydantic模型生成器。

添加新功能

添加新功能时，请遵循以下步骤：

在 src/api/ 下创建新的功能模块目录（如 users/）

创建模块的 __init__.py 文件，定义模块的公共API：

"""
用户管理模块 (Users Module)

负责用户账户的创建、查询、更新和删除功能。
"""

from .endpoints import router
from .schemas import UserCreate, UserResponse, UserList

__all__ = [
    "router",
    "UserCreate",
    "UserResponse",
    "UserList"
]

在模块目录中添加必要的文件：

constants.py: 定义模块相关常量和错误代码

"""用户模块常量定义。"""

# 错误消息
USER_NOT_FOUND_MSG = "找不到指定的用户"
DUPLICATE_EMAIL_MSG = "邮箱地址已被使用"

# 分页默认值
DEFAULT_PAGE_SIZE = 20
MAX_PAGE_SIZE = 100

exceptions.py: 定义模块特定的异常类

"""用户模块特定异常。"""

from fastapi import status
from src.core.exceptions import APIError

class UserNotFoundError(APIError):
    """用户未找到异常。"""
    
    def __init__(self, user_id: str, message: str = "找不到指定的用户"):
        super().__init__(
            message=message,
            status_code=status.HTTP_404_NOT_FOUND,
            details={"user_id": user_id}
        )

schemas.py: 定义请求和响应数据模型

"""用户模块数据模型。"""

from typing import List, Optional
from pydantic import Field, EmailStr

from src.core.base_models import CustomBaseModel, DataResponse, ListResponse

class UserBase(CustomBaseModel):
    """用户基本信息模型。"""
    name: str = Field(..., description="用户名称")
    email: EmailStr = Field(..., description="用户邮箱")

class UserCreate(UserBase):
    """用户创建请求模型。"""
    password: str = Field(..., description="用户密码")

class User(UserBase):
    """用户信息模型。"""
    id: int = Field(..., description="用户ID")
    is_active: bool = Field(True, description="是否激活")
    created_at: str = Field(..., description="创建时间")

class UserResponse(DataResponse[User]):
    """用户响应模型。"""
    pass

class UserList(ListResponse[User]):
    """用户列表响应模型。"""
    pass

service.py: 实现业务逻辑

"""用户模块业务逻辑。"""

from typing import List, Optional
from loguru import logger

from src.api.users.exceptions import UserNotFoundError
from src.db.models.user import User

async def get_user_by_id(user_id: int) -> User:
    """根据ID获取用户。"""
    user = await User.filter(id=user_id).first()
    if not user:
        logger.warning(f"找不到ID为{user_id}的用户")
        raise UserNotFoundError(user_id=str(user_id))
    return user

async def create_new_user(name: str, email: str, password: str) -> User:
    """创建新用户。"""
    # 实际实现中应该做一些验证工作
    user = await User.create(
        name=name,
        email=email,
        password_hash=password  # 在实际实现中应该先哈希
    )
    logger.info(f"已创建新用户: {user.name} (ID: {user.id})")
    return user

endpoints.py: 定义API路由和处理函数

"""用户模块路由定义。"""

from fastapi import APIRouter, Depends, status
from loguru import logger

from src.api.users.schemas import UserCreate, UserResponse, User, UserList
from src.api.users.service import get_user_by_id, create_new_user
from src.api.dependencies import get_current_developer_id

router = APIRouter()

@router.post(
    "", 
    response_model=UserResponse, 
    status_code=status.HTTP_201_CREATED,
    summary="创建新用户"
)
async def create_user(user_data: UserCreate, developer_id: str = Depends(get_current_developer_id)):
    """创建新用户。"""
    logger.info(f"开发者 {developer_id} 请求创建新用户")
    
    user = await create_new_user(
        name=user_data.name,
        email=user_data.email,
        password=user_data.password
    )
    
    return UserResponse(
        data=User(
            id=user.id,
            name=user.name,
            email=user.email,
            is_active=user.is_active,
            created_at=user.created_at.isoformat()
        )
    )

如果需要数据库模型，在 src/db/models/ 下创建相应的模型文件：

# src/db/models/user.py
from tortoise import fields
from tortoise.contrib.pydantic import pydantic_model_creator

from .base import AbstractBaseModel

class User(AbstractBaseModel):
    """用户模型。"""
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=255, unique=True)
    password_hash = fields.CharField(max_length=255)
    is_active = fields.BooleanField(default=True)
    
    class Meta:
        table = "users"
        ordering = ["name"]
    
    def __str__(self) -> str:
        return self.name

# Pydantic模型生成器
User_Pydantic = pydantic_model_creator(User, name="User")
UserIn_Pydantic = pydantic_model_creator(User, name="UserIn", exclude_readonly=True, exclude=["password_hash"])

在 src/db/config.py 的 MODELS_MODULES 列表中添加新模型：

# src/db/config.py
MODELS_MODULES: List[str] = [
    "src.db.models.api_key",
    "src.db.models.developer",
    "src.db.models.base",
    "src.db.models.user",  # 新模型
]

在 src/api/router.py 中注册新模块的路由：

# src/api/router.py
from fastapi import APIRouter
from src.api.auth import router as auth_router
from src.api.health import router as health_router
from src.api.users import router as users_router  # 导入新模块的路由

api_router = APIRouter()

# 注册路由
api_router.include_router(auth_router, prefix="/auth", tags=["auth"])
api_router.include_router(health_router, prefix="/health", tags=["health"])
api_router.include_router(users_router, prefix="/users", tags=["users"])  # 注册新路由

生成并应用数据库迁移：

# 生成迁移文件
python -m src.cli.main db migrate "添加用户模型"

# 应用迁移
python -m src.cli.main db upgrade

代码规范与质量

项目使用以下工具保证代码质量：

Ruff - 集成的代码检查和格式化工具
mypy - 静态类型检查

安装 pre-commit 钩子：

pre-commit install

数据库迁移

项目集成了 Aerich 进行数据库迁移管理，并提供了便捷的 CLI 工具。

使用 CLI 工具管理数据库迁移

生成迁移文件：

python -m src.cli.main db migrate "您的迁移描述"

应用迁移：

python -m src.cli.main db upgrade

回滚迁移：

python -m src.cli.main db downgrade  # 回滚到上一个版本
# 或者指定版本
# python -m src.cli.main db downgrade "0.1.0"

查看迁移历史：

python -m src.cli.main db history

查看数据库模型结构：

python -m src.cli.main db show-models

异步优先

本框架采用"异步优先"原则，尽可能使用 async def 定义路由和依赖，特别是在涉及 I/O 操作时。框架提供的所有依赖和服务都设计为异步友好的。

异步设计原则：

路由处理函数保持简洁：路由处理函数主要负责HTTP请求/响应处理，业务逻辑委托给服务层
服务层实现异步业务逻辑：实现异步业务逻辑，例如数据库查询、外部API调用等
异步异常处理：在异步函数中正确处理异常，确保异步上下文正确关闭

示例：

# endpoints.py
@router.get("/users/{user_id}")
async def get_user(user_id: int, developer_id: str = Depends(get_current_developer_id)):
    # 调用服务层的异步函数
    user = await get_user_by_id(user_id)
    return UserResponse(data=user)

# service.py
async def get_user_by_id(user_id: int) -> User:
    # 异步数据库查询
    user = await User.filter(id=user_id).first()
    if not user:
        raise UserNotFoundError(user_id=str(user_id))
    return user

模块化异常处理

本框架采用模块化的异常处理系统，每个模块都定义了特定的异常类，继承自全局基础异常类。这种设计允许更精确的错误处理和响应。

异常层次结构

基础异常：src.core.exceptions.APIError
- 定义了基本的异常属性：消息、状态码和详细信息
认证异常：src.core.exceptions.AuthenticationError
- 专门用于认证失败的异常，包含特定的HTTP头信息
模块特定异常：每个模块都定义自己的异常类
- src.api.auth.exceptions：定义了 InvalidApiKeyError, DeveloperNotFoundError 等
- src.api.health.exceptions：定义了 HealthCheckError, DatabaseConnectionError 等

异常使用示例

# 在服务层抛出特定异常
async def get_api_key_by_key(api_key_value: str) -> ApiKey:
    api_key = await ApiKey.filter(key=api_key_value).first()
    if not api_key:
        raise InvalidApiKeyError(message="无效的API密钥")
    return api_key

# 在路由处理函数中捕获异常
@router.post("/token")
async def login_for_access_token(form_data: TokenRequest):
    try:
        # 调用服务层函数
        access_token = await authenticate_api_key(
            api_key=form_data.api_key,
            api_secret=form_data.api_secret
        )
        return TokenResponse(data=Token(access_token=access_token))
    except InvalidApiKeyError:
        # 特定异常直接重新抛出
        raise
    except Exception as e:
        # 其他异常包装为APIError
        logger.exception(f"认证错误: {str(e)}")
        raise APIError(message="认证处理失败")

全局异常处理

FastAPI允许注册异常处理器，统一处理特定类型的异常：

@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "message": exc.message,
            "details": exc.details,
            "status": "error"
        }
    )

🧪 测试

运行测试：

# 启动测试数据库
docker run -p 3306:3306 -e MYSQL_ROOT_PASSWORD=test -e MYSQL_DATABASE=test_db -d mysql:8.4

# 运行测试
pytest -vvs tests/

📦 部署

生产环境变量

创建 .env.prod 文件并配置生产环境变量：

ENVIRONMENT=production
DEBUG=False
DB_PASSWORD=your_secure_production_password
SECRET_KEY=your_secure_secret_key
JWT_SECRET=your_secure_jwt_secret
# 其他生产环境配置...

使用 Docker Compose 部署

docker-compose -f docker-compose.yml -f deploy/docker-compose.prod.yml up -d

使用 Gunicorn 运行

gunicorn src.gunicorn_runner:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

🔒 安全最佳实践

在生产环境中使用强密码和密钥
使用 HTTPS 加密传输
定期轮换 API 密钥和 JWT 密钥
实施速率限制以防止滥用
对所有敏感操作使用认证和授权

📋 FastAPI 最佳实践

本框架遵循 FastAPI 最佳实践：

模块化路由：按功能域组织代码
依赖注入：使用依赖系统实现认证和参数验证
Pydantic 模型：为请求和响应定义清晰的数据模型
异步处理：利用异步特性提高性能
文档自动生成：利用 FastAPI 的自动文档功能

🤝 贡献

欢迎贡献代码、报告问题或提出改进建议。请确保遵循项目的代码规范和贡献指南。

FilesExpand file tree

README_zh.md

Latest commit

History

README_zh.md

File metadata and controls

FastAPI 启动开发框架

✨ 功能特性

🚀 快速开始

前提条件

安装依赖

配置环境变量

初始化数据库

运行开发服务器

🐳 使用 Docker 运行

开发模式

生产模式

🛠️ CLI 工具使用指南

基本使用

开发者管理命令

API 密钥管理命令

数据库管理命令

🔧 Aerich 数据迁移使用指南

初始化数据库

方式一：使用初始化脚本

方式二：使用 CLI 命令逐步操作

创建新的数据库迁移

管理数据库迁移

查看迁移历史

回滚迁移

查看数据库模型结构

Aerich 配置文件

注意事项

📚 API 文档

主要接口

认证

健康检查

🛠️ 开发指南

项目结构

按功能域组织代码

添加新功能

代码规范与质量

数据库迁移

使用 CLI 工具管理数据库迁移

异步优先

模块化异常处理

异常层次结构

异常使用示例

全局异常处理

🧪 测试

📦 部署

生产环境变量

使用 Docker Compose 部署

使用 Gunicorn 运行

🔒 安全最佳实践

📋 FastAPI 最佳实践

🤝 贡献