FastAPI三层架构：多服务协同处理复杂接口

在 FastAPI 中实现三层架构，特别是处理需要多个服务协同的复杂端点时，如何有效地组织代码至关重要。本文将深入探讨两种方案，并提供选择合适方案的指导，以实现更好的可维护性和可扩展性。

三层架构概述

三层架构是一种常见的软件设计模式，它将应用程序分为三个逻辑层：

• 表示层（Presentation Layer）： 负责用户交互，例如 FastAPI 中的端点定义。
• 应用层（Application Layer）： 处理业务逻辑，协调不同服务。
• 数据层（Data Layer）： 负责数据存储和访问，通常由各种服务提供。

在处理复杂端点时，我们需要仔细考虑如何在这些层之间分配职责，以确保代码的清晰性和可维护性。

方案一：应用层直接调用多个服务

这种方案将多个服务的调用逻辑放在应用层（即 FastAPI 端点定义处）。例如，对于 get_transaction 端点，应用层将直接调用 userService、productService 和 saleService 来获取所需的数据，然后将这些数据组合成 transactionDto 对象并返回。

示例代码：

from fastapi import FastAPI, Depends
from typing import Dict

app = FastAPI()

# 假设的 Service 定义
class UserService:
    def get_user(self, user_id: int) -> Dict:
        # 模拟数据库查询
        return {"user_id": user_id, "name": "Example User"}

class ProductService:
    def get_product(self, product_id: int) -> Dict:
        # 模拟数据库查询
        return {"product_id": product_id, "name": "Example Product"}

class SaleService:
    def get_sale(self, sale_id: int) -> Dict:
        # 模拟数据库查询
        return {"sale_id": sale_id, "amount": 100}

# 依赖注入
def get_user_service():
    return UserService()

def get_product_service():
    return ProductService()

def get_sale_service():
    return SaleService()

# 端点定义
@app.get("/transactions/{transaction_id}")
async def get_transaction(
    transaction_id: int,
    user_service: UserService = Depends(get_user_service),
    product_service: ProductService = Depends(get_product_service),
    sale_service: SaleService = Depends(get_sale_service),
):
    user = user_service.get_user(1)
    product = product_service.get_product(1)
    sale = sale_service.get_sale(transaction_id)

    transaction = {
        "transaction_id": transaction_id,
        "user": user,
        "product": product,
        "sale": sale,
    }

    return transaction

优点：

• 简单直接，易于理解和实现。
• 避免了服务之间的循环依赖。

缺点：

• 应用层承担了过多的业务逻辑，可能导致代码臃肿。
• 如果多个端点都需要相同的数据聚合逻辑，则会存在代码重复。

方案二：创建专门的聚合服务

这种方案创建一个专门的 transactionService 来聚合来自其他服务的数据。transactionService 将调用 userService、productService 和 saleService，并将它们的数据组合成 transaction 对象，然后将该对象返回给应用层。

示例代码：

from fastapi import FastAPI, Depends
from typing import Dict

app = FastAPI()

# 假设的 Service 定义
class UserService:
    def get_user(self, user_id: int) -> Dict:
        # 模拟数据库查询
        return {"user_id": user_id, "name": "Example User"}

class ProductService:
    def get_product(self, product_id: int) -> Dict:
        # 模拟数据库查询
        return {"product_id": product_id, "name": "Example Product"}

class SaleService:
    def get_sale(self, sale_id: int) -> Dict:
        # 模拟数据库查询
        return {"sale_id": sale_id, "amount": 100}

class TransactionService:
    def __init__(self, user_service: UserService, product_service: ProductService, sale_service: SaleService):
        self.user_service = user_service
        self.product_service = product_service
        self.sale_service = sale_service

    def get_transaction(self, transaction_id: int) -> Dict:
        user = self.user_service.get_user(1)
        product = self.product_service.get_product(1)
        sale = self.sale_service.get_sale(transaction_id)

        transaction = {
            "transaction_id": transaction_id,
            "user": user,
            "product": product,
            "sale": sale,
        }
        return transaction

# 依赖注入
def get_user_service():
    return UserService()

def get_product_service():
    return ProductService()

def get_sale_service():
    return SaleService()

def get_transaction_service(
    user_service: UserService = Depends(get_user_service),
    product_service: ProductService = Depends(get_product_service),
    sale_service: SaleService = Depends(get_sale_service),
):
    return TransactionService(user_service, product_service, sale_service)

# 端点定义
@app.get("/transactions/{transaction_id}")
async def get_transaction(
    transaction_id: int,
    transaction_service: TransactionService = Depends(get_transaction_service),
):
    transaction = transaction_service.get_transaction(transaction_id)
    return transaction

优点：

• 将数据聚合逻辑集中在一个地方，避免了代码重复。
• 降低了应用层的复杂性，使其更专注于处理请求和响应。
• 可以更容易地对数据聚合逻辑进行测试和维护。

缺点：

• 增加了服务的数量，可能增加部署和管理的复杂性。
• 服务之间的调用可能导致性能瓶颈。

如何选择合适的方案

选择哪种方案取决于具体的应用场景和需求。以下是一些考虑因素：

• 数据聚合逻辑的复杂性： 如果数据聚合逻辑非常复杂，建议使用方案二，创建一个专门的聚合服务。
• 代码重复： 如果多个端点都需要相同的数据聚合逻辑，建议使用方案二，以避免代码重复。
• 性能： 如果服务之间的调用可能导致性能瓶颈，需要仔细评估方案二的性能影响。
• 服务身份： 考虑服务是否有自己的存储和身份。如果服务有自己的存储和身份，那么它可能更适合作为一个独立的 RESTful 服务。如果没有，它可能更适合作为 BFF (Backend For Frontend) 或聚合服务。

总结：

两种方案都有其优缺点，选择哪种方案取决于具体的应用场景和需求。在选择方案时，需要仔细评估各种因素，并权衡其优缺点，以选择最适合的方案。重要的是保持代码的清晰性和可维护性，并确保应用程序的性能和可扩展性。