FastAPI：SOLID原则与设计模式

近年来，FastAPI凭借其简洁性、速度以及对静态类型的支持，已成为构建Python API的热门框架之一。然而，要充分利用这一强大工具，遵循一些良好的实践和设计模式至关重要，这些模式可以帮助我们编写出清晰、可维护且可扩展的代码。

在本文中，我将向您展示如何将一些SOLID原则和设计模式（如DAO（数据访问对象）、服务层和依赖注入）应用于FastAPI，以构建健壮且高效的API。

SOLID原则应用于FastAPI

1. 单一职责原则（SRP）

每个模块或类应该仅负责软件提供的功能的一部分，并且该职责应完全封装在类中。

这是什么意思呢？例如，在FastAPI应用程序中，路由函数（端点）应该专注于接收请求，将业务逻辑委托给特定的服务，并返回响应。我们来看一个代码示例：

违反SRP的代码：

from fastapi import APIRouter
from app.models.user import UserCreate, UserRead
from app.db import database

router = 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, UserDB
from app.db import database

class 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, UserRead
from app.repositories.user_repository import UserRepository

class 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, Depends
from app.models.user import UserCreate, UserRead
from app.services.user_service import UserService
from app.routers.dependencies import get_user_service

router = 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），不仅可以使代码更清晰、更易于维护，还可以为应用程序的未来发展奠定坚实的基础。

2. 依赖倒置原则（DIP）

该原则规定了两条规则：

1. 高层模块不应依赖于低层模块。两者都应依赖于抽象。
2. 抽象不应依赖于细节。细节必须依赖于抽象。

回到之前的例子：UserService 类直接依赖于 UserRepository 的具体实现。这是一种违反DIP的行为，因为 UserService（高层模块）依赖于 UserRepository（低层模块）。如果明天您决定改变 UserRepository 处理数据持久化的方式（例如，从SQL迁移到NoSQL），您还将不得不修改 UserService。

要正确应用DIP，我们需要引入一个抽象（接口或基类），它定义了 UserService 与用户仓库交互所需的操作契约。这样，UserService 将依赖于抽象，而不是具体实现。

我们将这个类称为 IUserRepository，表示它是一个接口：

from abc import ABC, abstractmethod
from app.models.user import UserCreate, UserRead

class 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, UserRead
from app.db import database
from app.repositories.user_repository_interface import IUserRepository

class 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, UserRead
from app.repositories.user_repository_interface import IUserRepository

class 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原则外，我们还使用了一些设计模式。您有没有发现是哪些呢？

1. 数据访问对象（DAO）模式

DAO模式是一种设计模式，用于将数据访问逻辑与应用程序的业务逻辑分离。它的目的是为在数据库或其他数据源上执行的CRUD（创建、读取、更新、删除）操作提供一个抽象层。

在我们的示例中，可以识别出UserRepository类。这个类将负责与用户实体相关的所有数据库交互。

from app.models.user import UserCreate, UserRead
from app.db import database
from app.repositories.user_repository_interface import IUserRepository

class 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})

为什么应该使用这个模式？

• 数据访问封装：DAO提供了一个专门的层来管理所有数据访问操作。这意味着任何持久化逻辑的变化（例如，从SQL切换到NoSQL）都只在DAO层进行，而不会影响应用程序的其他部分。
• 可重用性：DAO的实现可以被需要与同一实体数据交互的不同服务或组件重用，从而避免代码重复。
• 易于测试：通过将数据访问单独放在一个层中，可以轻松创建模拟对象或桩对象以进行单元测试，从而允许在不依赖实际数据库的情况下独立测试业务逻辑。
• 可维护性：数据访问操作集中在DAO类中，便于定位和修正与持久化相关的错误。

2. 服务层

服务层负责组织应用程序的业务逻辑。它的目的是为表示层或控制器（例如FastAPI控制器）提供一个接口，封装所有相关的业务逻辑。

通过将业务逻辑分离到服务层，可以实现更模块化、更可维护且可测试的代码。此外，它还便于在不同的应用程序上下文中重用业务逻辑。

在我们的示例中，我们在UserService类中使用了这个模式。这个类负责与用户相关的业务逻辑。这些业务逻辑可能包括验证、业务规则、数据转换等。它将使用DAO来执行数据访问操作，但不会直接管理数据库。

from app.models.user import UserCreate, UserRead
from app.repositories.user_repository_interface import IUserRepository

class 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

为什么应该使用这个模式？

• 业务逻辑分离：为所有业务逻辑提供了一个集中的位置，消除了应用程序不同部分中的重复逻辑，并将其与数据访问逻辑分开。
• 易于单元测试：通过将业务逻辑封装在单独的服务层中，可以更容易地对业务逻辑进行单元测试，而不必担心数据库的细节。这允许创建不同的模拟对象来测试不同的场景。
• 代码解耦：表示层（例如FastAPI中的控制器）不会直接依赖于业务逻辑或数据访问层。这允许在不影响用户界面的情况下更改业务逻辑，反之亦然。
• 灵活性和可扩展性：作为业务逻辑的中心点，可以在不影响应用程序其他部分的情况下，应用额外的更改、规则或验证。

总结

通过遵循这些设计原则和模式，您可以使用FastAPI构建出更健壮、灵活且可维护的API。通过应用SOLID原则以及DAO或服务层等设计模式，您不仅可以提高代码的质量，还可以增强其适应变化和随着时间推移而发展的能力。