摘要：近年来，FastAPI凭借其简洁性、速度以及对静态类型的支持，已成为构建Python API的热门框架之一。然而，要充分利用这一强大工具，遵循一些良好的实践和设计模式至关重要，这些模式可以帮助我们编写出清晰、可维护且可扩展的代码。
近年来，FastAPI凭借其简洁性、速度以及对静态类型的支持，已成为构建Python API的热门框架之一。然而，要充分利用这一强大工具，遵循一些良好的实践和设计模式至关重要，这些模式可以帮助我们编写出清晰、可维护且可扩展的代码。
在本文中，我将向您展示如何将一些SOLID原则和设计模式（如DAO（数据访问对象）、服务层和依赖注入）应用于FastAPI，以构建健壮且高效的API。
每个模块或类应该仅负责软件提供的功能的一部分，并且该职责应完全封装在类中。
这是什么意思呢？例如，在FastAPI应用程序中，路由函数（端点）应该专注于接收请求，将业务逻辑委托给特定的服务，并返回响应。我们来看一个代码示例：
违反SRP的代码：
from fastapi import APIRouterfrom app.models.user import UserCreate, UserReadfrom app.db import databaserouter = APIRouter@router.post("/users", response_model=UserRead)async def create_user(user: UserCreate): if not user.email or not user.password: raise ValueError("Email and password are required.") existing_user = database.fetch_one("SELECT * FROM users WHERE email = :email", {"email": user.email}) if existing_user: raise ValueError("User already exists.") new_user_id = database.execute("INSERT INTO users (email, password) VALUES (:email, :password)", { "email": user.email, "password": user.password }) new_user = database.fetch_one("SELECT * FROM users WHERE id = :id", {"id": new_user_id}) return new_user在这个例子中，创建用户端点承担了多个任务：
验证输入数据（验证邮箱和密码）。检查用户是否已存在于数据库中。在数据库中创建新用户。检索并返回新用户的信息。这种职责的混合使得代码更难以维护和扩展。业务逻辑或数据访问方式的任何变化都需要修改这段代码，从而增加了出错的可能性。
遵循SRP的代码：
from app.models.user import UserCreate, UserDBfrom app.db import databaseclass UserRepository: def __init__(self, db_session): self.db_session = db_session async def get_user_by_email(self, email: str) -> UserDB: query = "SELECT * FROM users WHERE email = :email" return await self.db_session.fetch_one(query, {"email": email}) async def add_user(self, user_data: UserCreate) -> int: query = "INSERT INTO users (email, password) VALUES (:email, :password) RETURNING id" values = {"email": user_data.email, "password": user_data.password} new_user_id = await self.db_session.execute(query, values) return new_user_id async def get_user_by_id(self, user_id: int) -> UserDB: query = "SELECT * FROM users WHERE id = :id" return await self.db_session.fetch_one(query, {"id": user_id})from app.models.user import UserCreate, UserReadfrom app.repositories.user_repository import UserRepositoryclass UserService: def __init__(self, user_repository: UserRepository): self.user_repository = user_repository async def validate_user_data(self, user_data: UserCreate) -> None: if not user_data.email or not user_data.password: raise ValueError("Email and password are required.") async def check_user_exists(self, email: str) -> None: existing_user = await self.user_repository.get_user_by_email(email) if existing_user: raise ValueError("User already exists.") async def create_user(self, user_data: UserCreate) -> UserRead: await self.validate_user_data(user_data) await self.check_user_exists(user_data.email) new_user_id = await self.user_repository.add_user(user_data) return await self.user_repository.get_user_by_id(new_user_id)from fastapi import APIRouter, Dependsfrom app.models.user import UserCreate, UserReadfrom app.services.user_service import UserServicefrom app.routers.dependencies import get_user_servicerouter = APIRouter@router.post("/users", response_model=UserRead)async def create_user(user: UserCreate, user_service: UserService = Depends(get_user_service)): return await user_service.create_user(user)现在，您可以清楚地看到每个模块的独特职责。
用户仓库（User Repository）：负责所有与数据库相关的操作，例如获取或插入用户。数据库结构或数据处理方式的任何变化都将在这里进行。用户服务（User Service）：包含与用户相关的业务逻辑，例如验证、业务规则等。它是仓库和FastAPI路由之间的中介。用户路由（User Router）：端点的功能非常简单：接收HTTP请求，将逻辑委托给相应的服务，并返回响应。它不关心数据或业务逻辑的内部细节。在FastAPI应用程序中应用单一职责原则（SRP），不仅可以使代码更清晰、更易于维护，还可以为应用程序的未来发展奠定坚实的基础。
高层模块不应依赖于低层模块。两者都应依赖于抽象。抽象不应依赖于细节。细节必须依赖于抽象。回到之前的例子：UserService 类直接依赖于 UserRepository 的具体实现。这是一种违反DIP的行为，因为 UserService（高层模块）依赖于 UserRepository（低层模块）。如果明天您决定改变 UserRepository 处理数据持久化的方式（例如，从SQL迁移到NoSQL），您还将不得不修改 UserService。
要正确应用DIP，我们需要引入一个抽象（接口或基类），它定义了 UserService 与用户仓库交互所需的操作契约。这样，UserService 将依赖于抽象，而不是具体实现。
我们将这个类称为 IUserRepository，表示它是一个接口：
from abc import ABC, abstractmethodfrom app.models.user import UserCreate, UserReadclass IUserRepository(ABC): @abstractmethod async def get_user_by_email(self, email: str) -> UserRead: pass @abstractmethod async def add_user(self, user_data: UserCreate) -> int: pass @abstractmethod async def get_user_by_id(self, user_id: int) -> UserRead: pass这个接口定义了任何用户仓库的具体实现都必须具备的方法。这确保了 UserService 可以与任何实现了该接口的仓库一起工作，而不依赖于它处理数据的方式。
现在，我们必须让 UserRepository 实现这个接口：
from app.models.user import UserCreate, UserReadfrom app.db import databasefrom app.repositories.user_repository_interface import IUserRepositoryclass UserRepository(IUserRepository): def __init__(self, db_session): self.db_session = db_session async def get_user_by_email(self, email: str) -> UserRead: query = "SELECT * FROM users WHERE email = :email" return await self.db_session.fetch_one(query, {"email": email}) async def add_user(self, user_data: UserCreate) -> int: query = "INSERT INTO users (email, password) VALUES (:email, :password) RETURNING id" values = {"email": user_data.email, "password": user_data.password} new_user_id = await self.db_session.execute(query, values) return new_user_id async def get_user_by_id(self, user_id: int) -> UserRead: query = "SELECT * FROM users WHERE id = :id" return await self.db_session.fetch_one(query, {"id": user_id})实现接口确保它提供了 UserService 可能需要的所有功能。
最后，我们必须让 UserService 类依赖于接口，而不是具体类：
from app.models.user import UserCreate, UserReadfrom app.repositories.user_repository_interface import IUserRepositoryclass UserService: def __init__(self, user_repository: IUserRepository): self.user_repository = user_repository async def validate_user_data(self, user_data: UserCreate) -> None: if not user_data.email or not user_data.password: raise ValueError("Email and password are required.") async def check_user_exists(self, email: str) -> None: existing_user = await self.user_repository.get_user_by_email(email) if existing_user: raise ValueError("User already exists.") async def create_user(self, user_data: UserCreate) -> UserRead: await self.validate_user_data(user_data) await self.check_user_exists(user_data.email) new_user_id = await self.user_repository.add_user(user_data) new_user = await self.user_repository.get_user_by_id(new_user_id) return new_user应用依赖倒置原则（DIP）通过解耦业务逻辑与实现细节，改善了应用程序架构。在这个例子中，UserService 通过依赖于 IUserRepository 接口而不是具体实现（如 UserRepository），变得更加灵活、可维护且易于测试。
在在前面的示例中，除了使用SOLID原则外，我们还使用了一些设计模式。您有没有发现是哪些呢？
DAO模式是一种设计模式，用于将数据访问逻辑与应用程序的业务逻辑分离。它的目的是为在数据库或其他数据源上执行的CRUD（创建、读取、更新、删除）操作提供一个抽象层。
在我们的示例中，可以识别出UserRepository类。这个类将负责与用户实体相关的所有数据库交互。
为什么应该使用这个模式？
数据访问封装：DAO提供了一个专门的层来管理所有数据访问操作。这意味着任何持久化逻辑的变化（例如，从SQL切换到NoSQL）都只在DAO层进行，而不会影响应用程序的其他部分。可重用性：DAO的实现可以被需要与同一实体数据交互的不同服务或组件重用，从而避免代码重复。易于测试：通过将数据访问单独放在一个层中，可以轻松创建模拟对象或桩对象以进行单元测试，从而允许在不依赖实际数据库的情况下独立测试业务逻辑。可维护性：数据访问操作集中在DAO类中，便于定位和修正与持久化相关的错误。服务层负责组织应用程序的业务逻辑。它的目的是为表示层或控制器（例如FastAPI控制器）提供一个接口，封装所有相关的业务逻辑。
通过将业务逻辑分离到服务层，可以实现更模块化、更可维护且可测试的代码。此外，它还便于在不同的应用程序上下文中重用业务逻辑。
在我们的示例中，我们在UserService类中使用了这个模式。这个类负责与用户相关的业务逻辑。这些业务逻辑可能包括验证、业务规则、数据转换等。它将使用DAO来执行数据访问操作，但不会直接管理数据库。
为什么应该使用这个模式？
业务逻辑分离：为所有业务逻辑提供了一个集中的位置，消除了应用程序不同部分中的重复逻辑，并将其与数据访问逻辑分开。易于单元测试：通过将业务逻辑封装在单独的服务层中，可以更容易地对业务逻辑进行单元测试，而不必担心数据库的细节。这允许创建不同的模拟对象来测试不同的场景。代码解耦：表示层（例如FastAPI中的控制器）不会直接依赖于业务逻辑或数据访问层。这允许在不影响用户界面的情况下更改业务逻辑，反之亦然。灵活性和可扩展性：作为业务逻辑的中心点，可以在不影响应用程序其他部分的情况下，应用额外的更改、规则或验证。通过遵循这些设计原则和模式，您可以使用FastAPI构建出更健壮、灵活且可维护的API。通过应用SOLID原则以及DAO或服务层等设计模式，您不仅可以提高代码的质量，还可以增强其适应变化和随着时间推移而发展的能力。
来源：散文随风想