FastAPI-Production-Boilerplate

FastAPI 生产环境模板

一个可扩展且适用于生产环境的 FastAPI 样板

目录

• 项目概述
• 特性
• 安装指南
• 使用指南
• 高级用法
• 贡献
• 许可证
• 致谢

项目概述

这个样板遵循分层架构，包括模型层、仓储层、控制器层和 API 层。其目录结构旨在将样板代码隔离在核心目录中，该目录只需最少关注，从而促进快速轻松的功能开发。目录结构通常也非常可预测。该项目的主要目标是提供一个生产就绪的样板，具有更好的开发者体验和现成可用的功能。它还包含一些广泛使用的功能，如身份验证、授权、数据库迁移、类型检查等，这些将在特性部分详细讨论。

特性

• 支持 Python 3.11+
• 支持 SQLAlchemy 2.0+
• 异步能力
• 使用 Alembic 进行数据库迁移
• 使用 JWT 进行基本身份验证
• 用于权限的行级访问控制
• 使用 Redis 进行缓存
• 使用 Celery 处理后台任务
• 测试套件
• 使用 mypy 进行类型检查
• Docker 化的数据库和 Redis
• 现成可用的 CRUD 操作
• 使用 pylint 进行代码检查
• 使用 black 进行代码格式化

安装指南

运行此项目需要以下条件：

• Python 3.11
• Docker 和 Docker Compose
• Poetry

我使用 asdf 来管理我的 Python 版本。你也可以使用它。但是，它只支持 Linux 和 macOS。对于 Windows，你可以使用类似 pyenv 的工具。

一旦你安装了上述工具并克隆了仓库，你可以按照以下步骤启动项目：

1. 使用 poetry 创建虚拟环境：

poetry shell

2. 安装依赖：

poetry install

3. 运行数据库和 Redis 容器：

docker-compose up -d

4. 将 .env.example 文件复制为 .env 并根据需要更新值。

5. 运行迁移：

make migrate

6. 运行服务器：

make run

现在服务器应该在 http://localhost:8000 上运行，API 文档应该在 http://localhost:8000/docs 上可用。

使用指南

该项目设计为模块化和可扩展的。项目中有 3 个主要目录：

1. core：这个目录包含项目的核心部分。它包含大部分样板代码，如安全依赖、数据库连接、配置、中间件等。它还包含模型、仓储和控制器的基类。core 目录设计得尽可能简约，通常只需最少关注。总的来说，core 目录设计得尽可能通用，可以用于任何项目。在构建额外功能时，你可能不需要修改这个目录，除非需要在 core/factory.py 中的 Factory 类中添加更多控制器。

2. app：这个目录包含实际的应用程序代码。它包含应用程序的模型、仓储、控制器和模式。在构建功能时，你将花费大部分时间在这个目录中。该目录有以下子目录：

   • models 这里是添加新表的地方
   • repositories 对于每个模型，你需要创建一个仓储。这是添加模型 CRUD 操作的地方。
   • controllers 对于应用程序的每个逻辑单元，你需要创建一个控制器。这是添加应用程序业务逻辑的地方。
   • schemas 这是添加应用程序模式的地方。模式用于验证和数据的序列化/反序列化。

3. api：这个目录包含应用程序的 API 层。它包含 API 路由器，这是添加 API 端点的地方。

高级用法

该样板包含许多功能，其中一些在应用程序中使用，一些没有。以下部分详细描述了这些功能。

数据库迁移

迁移由 Alembic 处理。迁移存储在 migrations 目录中。要创建新的迁移，可以运行以下命令：

make generate-migration

它会要求你为迁移输入一条消息。输入消息后，它会在 migrations 目录中创建一个新的迁移文件。然后你可以使用以下命令运行迁移：

make migrate

如果你需要降级数据库或重置它。你可以分别使用 make rollback 和 make reset-database。

身份验证

使用的身份验证是 JWT 的基本实现，带有承载令牌。当在 Authorization 头中提供 bearer 令牌时，令牌会被验证，用户会通过中间件自动认证，设置 request.user.id。要在任何端点中使用用户模型，可以使用 get_current_user 依赖。如果你想在任何端点强制认证，可以使用 AuthenticationRequired 依赖。如果用户未认证，它会抛出 HTTPException。

行级访问控制

该样板包含一个自定义的行级权限管理模块。它受到 fastapi-permissions 的启发。它位于 core/security/access_control.py。你可以使用它为不同的模型强制执行不同的权限。该模块基于 Principals 和 permissions 运作。每个用户都有自己的一组principals，需要使用函数设置。查看 core/fastapi/dependencies/permissions.py 以获取示例。然后使用principals来检查用户的权限。需要在模型级别定义权限。查看 app/models/user.py 以获取示例。然后你可以直接在路由中使用依赖项，如果用户没有所需的权限，就会抛出 HTTPException。以下是一个不完整的示例：

from fastapi import APIRouter, Depends
from core.security.access_control import AccessControl, UserPrincipal, RolePrincipal, Allow
from core.database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)
    password = Column(String)
    role = Column(String)
def __acl__(self):
    return [
        (Allow, UserPrincipal(self.id), "view"),
        (Allow, RolePrincipal("admin"), "delete"),
    ]

def get_user_principals(user: User = Depends(get_current_user)):
    return [UserPrincipal(user.id)]

Permission = AccessControl(get_user_principals)

router = APIRouter()

@router.get("/users/{user_id}")
def get_user(user_id: int, user: User = get_user(user_id), assert_access = Permission("view")):
    assert_access(user)
    return user

缓存

你可以直接使用来自 core.cache 的 Cache.cached 装饰器。示例：

from core.cache import Cache

@Cache.cached(prefix="user", ttl=60)
def get_user(user_id: int):
    ...

Celery

Celery 工作进程已为应用程序配置好。你可以在 worker/ 中添加你的任务。要运行 Celery 工作进程，你可以运行以下命令：

make celery-worker

会话管理

会话已由中间件和 get_session 依赖项处理，该依赖项通过 FastAPI 依赖注入在 core/factory.py 中的 Factory 类中注入到仓库中。还有一个 Transactional 装饰器，可用于包装需要在事务中执行的函数。示例：

@Transactional()
async def some_mutating_function():
    ...

注意：该装饰器已经处理了事务的提交和回滚。你不需要手动执行。

如果在任何情况下你需要独立的会话，你可以使用 core.database 中的 standalone_session 装饰器。示例：

@standalone_session
async def do_something():
    ...

仓库模式

该样板使用仓库模式。每个模型都有一个仓库，所有仓库都继承自 core/repository 中的 base 仓库。仓库位于 app/repositories 中。仓库在 core/factory/factory.py 中的 Factory 类中注入到控制器中。

基础仓库具有基本的 CRUD 操作。所有自定义操作都可以添加到特定的仓库中。示例：

from core.repository import BaseRepository
from app.models.user import User
from sqlalchemy.sql.expression import select

class UserRepository(BaseRepository[User]):
    async def get_by_email(self, email: str):
        return await select(User).filter(User.email == email).gino.first()

为了更方便地访问复杂连接的查询，BaseRepository 类有一个 _query 函数（以及其他方便的函数，如 _all() 和 _one_or_none()），可以用来非常轻松地编写复杂查询。示例：

async def get_user_by_email_join_tasks(email: str):
    query = await self._query(join_)
    query = query.filter(User.email == email)
    return await self._one_or_none(query)

注意：对于每个你想进行的连接，你需要在同一个仓库中创建一个名称模式为 _join_{name} 的函数。例如：tasks 的 _join_tasks。示例：

async def _join_tasks(self, query: Select) -> Select:
    return query.options(joinedload(User.tasks))

控制器

与仓库类似，应用程序的每个逻辑单元都有一个控制器。控制器还有一个主要仓库，该仓库被注入其中。控制器位于 app/controllers 中。

这些控制器包含应用程序的所有业务逻辑。查看 app/controllers/auth.py 以获取示例。

架构

架构位于 app/schemas 中。架构用于验证请求体和响应体。架构还用于生成 OpenAPI 文档。架构继承自 pydantic 的 BaseModel。架构主要被隔离为 requests 和 responses，这两者的含义不言自明。

格式化

你可以使用 make format 来使用 black 和 isort 格式化代码。

代码检查

你可以使用 make lint 来使用 pylint 检查代码。

测试

该项目包含所有端点的测试，一些逻辑组件如 JWTHander 和 AccessControl 的测试，以及测试复杂内部组件如 BaseRepository 的示例。测试位于 tests/ 中。你可以使用 make test 运行测试。

贡献

非常欢迎贡献。如果你想贡献，请开一个 issue 或提交一个 PR。

许可证

本项目根据 MIT 许可证的条款授权。请查看 LICENSE 文件。

致谢

• 本项目使用了来自 teamhide/fastapi-boilerplate 的几个组件
• 行级访问控制受 fastapi-permissions 启发
• CRUD 模式受 full-stack-fastapi-postgresql 启发