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 的工具。
一旦你安装了上述工具并克隆了仓库,你可以按照以下步骤启动项目:
- 使用 poetry 创建虚拟环境:
poetry shell
- 安装依赖:
poetry install
- 运行数据库和 Redis 容器:
docker-compose up -d
-
将
.env.example
文件复制为.env
并根据需要更新值。 -
运行迁移:
make migrate
- 运行服务器:
make run
现在服务器应该在 http://localhost:8000
上运行,API 文档应该在 http://localhost:8000/docs
上可用。
使用指南
该项目设计为模块化和可扩展的。项目中有 3 个主要目录:
-
core
:这个目录包含项目的核心部分。它包含大部分样板代码,如安全依赖、数据库连接、配置、中间件等。它还包含模型、仓储和控制器的基类。core
目录设计得尽可能简约,通常只需最少关注。总的来说,core
目录设计得尽可能通用,可以用于任何项目。在构建额外功能时,你可能不需要修改这个目录,除非需要在core/factory.py
中的Factory
类中添加更多控制器。 -
app
:这个目录包含实际的应用程序代码。它包含应用程序的模型、仓储、控制器和模式。在构建功能时,你将花费大部分时间在这个目录中。该目录有以下子目录:models
这里是添加新表的地方repositories
对于每个模型,你需要创建一个仓储。这是添加模型 CRUD 操作的地方。controllers
对于应用程序的每个逻辑单元,你需要创建一个控制器。这是添加应用程序业务逻辑的地方。schemas
这是添加应用程序模式的地方。模式用于验证和数据的序列化/反序列化。
-
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 启发