在 FastAPI 中实现三层架构时，处理需要多个服务支持的复杂 Endpoint 的最佳实践。针对诸如“get_transaction”这类需要聚合用户、产品和销售数据的情况，分析了在应用层直接调用多个服务，还是创建一个专门的聚合服务两种方案的优劣，并提出了基于服务身份和存储的拆分策略建议，以提升系统的可扩展性和可维护性。
在构建复杂的 FastAPI 应用时，经常会遇到一个 Endpoint 需要从多个不同的服务获取数据才能完成请求的情况。例如，一个“get_transaction” Endpoint 可能需要从用户服务、产品服务和销售服务获取数据，然后将这些数据整合为一个“transactionDto”对象返回给客户端。面对这种情况，如何设计服务间的交互，才能保证代码的清晰、可维护性和可扩展性，是一个值得深入探讨的问题。
方案一：应用层直接调用多个服务
在这种方案中，Endpoint 的逻辑直接位于应用层（通常是 FastAPI 的 router 中）。应用层负责调用用户服务、产品服务和销售服务，获取各自的数据，然后将这些数据组合成 transactionDto 对象。
from fastapi import APIRouter, Depends
from .services import UserService, ProductService, SaleService
from .schemas import TransactionDto
router = APIRouter()
@router.get("/transactions/{transaction_id}")
async def get_transaction(
transaction_id: int,
user_service: UserService = Depends(),
product_service: ProductService = Depends(),
sale_service: SaleService = Depends(),
) -> TransactionDto:
user = await user_service.get_user_by_transaction_id(transaction_id)
product = await product_service.get_product_by_transaction_id(transaction_id)
sale = await sale_service.get_sale_by_transaction_id(transaction_id)
transaction_dto = TransactionDto(
user=user,
product=product,
sale=sale,
)
return transaction_dto优点：
- 简单直接，易于理解。
- 减少了服务间的调用链，降低了延迟。
缺点：
- 应用层承担了过多的业务逻辑，违反了单一职责原则。
- 如果 transactionDto 的构建逻辑复杂，会导致应用层代码臃肿。
- 如果多个 Endpoint 都需要相同的聚合逻辑，会产生代码重复。
方案二：创建专门的聚合服务
在这种方案中，创建一个名为 TransactionService 的服务，专门负责聚合来自用户服务、产品服务和销售服务的数据。应用层只需要调用 TransactionService，即可获取完整的 transactionDto 对象。
from fastapi import APIRouter, Depends
from .services import TransactionService
from .schemas import TransactionDto
router = APIRouter()
@router.get("/transactions/{transaction_id}")
async def get_transaction(
transaction_id: int,
transaction_service: TransactionService = Depends(),
) -> TransactionDto:
transaction_dto = await transaction_service.get_transaction(transaction_id)
return transaction_dtoTransactionService 的实现：
from .services import UserService, ProductService, SaleService
from .schemas import TransactionDto
class TransactionService:
def __init__(
self,
user_service: UserService = Depends(),
product_service: ProductService = Depends(),
sale_service: SaleService = Depends(),
):
self.user_service = user_service
self.product_service = product_service
self.sale_service = sale_service
async def get_transaction(self, transaction_id: int) -> TransactionDto:
user = await self.user_service.get_user_by_transaction_id(transaction_id)
product = await self.product_service.get_product_by_transaction_id(transaction_id)
sale = await self.sale_service.get_sale_by_transaction_id(transaction_id)
transaction_dto = TransactionDto(
user=user,
product=product,
sale=sale,
)
return transaction_dto优点：
- 应用层代码简洁，只负责处理 HTTP 请求和响应。
- 聚合逻辑集中在 TransactionService 中，易于维护和测试。
- 可以复用 TransactionService，避免代码重复。
缺点：
- 增加了服务间的调用链，可能导致延迟增加。
- 需要考虑服务间的依赖关系和版本兼容性。
服务拆分策略：基于身份和存储
选择哪种方案，取决于具体的业务场景和需求。一般来说，如果聚合逻辑比较复杂，或者多个 Endpoint 都需要相同的聚合逻辑，那么创建专门的聚合服务是一个更好的选择。
更进一步，可以考虑基于服务的身份和存储来决定是否应该创建独立的聚合服务。如果聚合后的数据具有独立的身份（例如，transactionDto 在系统中作为一个独立的实体存在），并且需要持久化存储，那么应该创建一个独立的聚合服务。这种服务更像是一个 Backend for Frontend (BFF) 或聚合服务，它负责将来自不同服务的的数据整合为一个完整的实体。
如果聚合后的数据只是临时性的，不需要持久化存储，那么也可以考虑在应用层直接调用多个服务。但需要注意，要尽量保持应用层代码的简洁和可读性，避免承担过多的业务逻辑。
注意事项：
- 服务间的通信方式： 可以选择 REST API、gRPC 或消息队列等方式进行服务间的通信。
- 服务发现： 在微服务架构中，需要使用服务发现机制来动态地查找和连接服务。
- 熔断和降级： 为了保证系统的稳定性，需要实现熔断和降级机制，防止服务雪崩。
- 监控和日志： 需要对服务进行监控和日志记录，以便及时发现和解决问题。
总结：
在 FastAPI 中实现三层架构处理复杂 Endpoint 时，需要根据具体的业务场景和需求选择合适的方案。创建专门的聚合服务可以提高代码的可维护性和可复用性，但也会增加服务间的调用链。基于服务的身份和存储来决定是否应该创建独立的聚合服务，是一个值得考虑的策略。同时，需要关注服务间的通信方式、服务发现、熔断和降级、监控和日志等方面，以保证系统的稳定性和可扩展性。
