在构建复杂的 FastAPI 应用时，采用三层架构（表现层、应用层、领域层）是一种常见的实践。然而，当某个 Endpoint 需要聚合来自多个不同服务的的数据时，例如一个 get_transaction Endpoint 需要用户、产品和销售信息，如何组织代码就成了一个需要仔细考虑的问题。常见的做法有两种：一种是在应用层直接调用多个服务，另一种是创建一个专门的服务来聚合这些数据。
方案一：应用层直接调用多个服务
在这种方案中，FastAPI Endpoint 直接调用用户服务、产品服务和销售服务，然后将返回的数据组装成 transactionDto 对象并返回。
from fastapi import FastAPI, Depends
from typing import Dict
# 假设的 User, Product, Sale 服务接口
class UserService:
async def get_user(self, user_id: int) -> Dict:
# 模拟从数据库或外部服务获取用户信息
return {"id": user_id, "name": "Example User"}
class ProductService:
async def get_product(self, product_id: int) -> Dict:
# 模拟从数据库或外部服务获取产品信息
return {"id": product_id, "name": "Example Product"}
class SaleService:
async def get_sale(self, sale_id: int) -> Dict:
# 模拟从数据库或外部服务获取销售信息
return {"id": sale_id, "amount": 100}
app = FastAPI()
@app.get("/transactions/{transaction_id}")
async def get_transaction(
transaction_id: int,
user_service: UserService = Depends(),
product_service: ProductService = Depends(),
sale_service: SaleService = Depends()
) -> Dict:
user = await user_service.get_user(transaction_id)
product = await product_service.get_product(transaction_id)
sale = await sale_service.get_sale(transaction_id)
transaction_data = {
"transaction_id": transaction_id,
"user": user,
"product": product,
"sale": sale
}
return transaction_data优点：
- 简单直接： 实现起来比较简单，逻辑清晰。
- 减少服务间的依赖： 应用层直接与多个服务交互，避免了服务之间的循环依赖。
缺点：
- 应用层逻辑复杂： 如果聚合逻辑复杂，会导致应用层代码臃肿，违反单一职责原则。
- 代码复用性差： 如果多个 Endpoint 都需要类似的聚合逻辑，则需要重复编写代码。
方案二：创建专门的 TransactionService
在这种方案中，创建一个 TransactionService，它负责调用用户服务、产品服务和销售服务，并将返回的数据组装成 transaction 对象，然后返回给应用层。
from fastapi import FastAPI, Depends
from typing import Dict
# 假设的 User, Product, Sale 服务接口
class UserService:
async def get_user(self, user_id: int) -> Dict:
# 模拟从数据库或外部服务获取用户信息
return {"id": user_id, "name": "Example User"}
class ProductService:
async def get_product(self, product_id: int) -> Dict:
# 模拟从数据库或外部服务获取产品信息
return {"id": product_id, "name": "Example Product"}
class SaleService:
async def get_sale(self, sale_id: int) -> Dict:
# 模拟从数据库或外部服务获取销售信息
return {"id": sale_id, "amount": 100}
# Transaction Service
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
async def get_transaction(self, transaction_id: int) -> Dict:
user = await self.user_service.get_user(transaction_id)
product = await self.product_service.get_product(transaction_id)
sale = await self.sale_service.get_sale(transaction_id)
transaction_data = {
"transaction_id": transaction_id,
"user": user,
"product": product,
"sale": sale
}
return transaction_data
app = FastAPI()
@app.get("/transactions/{transaction_id}")
async def get_transaction(
transaction_id: int,
transaction_service: TransactionService = Depends()
) -> Dict:
transaction = await transaction_service.get_transaction(transaction_id)
return transaction优点：
- 职责明确： TransactionService 专门负责聚合数据，应用层代码更简洁。
- 代码复用性高： 如果多个 Endpoint 都需要类似的聚合逻辑，可以复用 TransactionService。
- 更容易进行单元测试： 可以针对 TransactionService 进行独立的单元测试。
缺点：
- 增加服务间的依赖： TransactionService 依赖于多个服务，可能导致服务之间的循环依赖。
- 可能增加延迟： 多了一层服务调用，可能会增加请求延迟。
如何选择？
选择哪种方案取决于具体的场景和需求。以下是一些建议：
- 聚合逻辑简单： 如果聚合逻辑非常简单，可以直接在应用层调用多个服务。
- 聚合逻辑复杂： 如果聚合逻辑复杂，或者多个 Endpoint 都需要类似的聚合逻辑，则应该创建一个专门的 TransactionService。
- 服务边界清晰： 仔细考虑每个服务的职责和边界，避免服务之间的循环依赖。
- 性能要求高： 如果对性能要求非常高，需要仔细评估增加 TransactionService 带来的延迟。
BFF (Backend For Frontend) 模式：
可以考虑将 TransactionService 视为一种 BFF (Backend For Frontend) 模式的应用。BFF 模式指的是为不同的前端应用创建不同的后端服务，这些后端服务负责聚合和转换数据，以满足前端应用的特定需求。
总结
在 FastAPI 中实现三层架构处理复杂 Endpoint 时，需要仔细考虑服务的设计和组织方式。应用层直接调用多个服务和创建专门的服务来聚合数据是两种常见的方案，每种方案都有其优缺点。选择哪种方案取决于具体的场景和需求。 需要注意的是，要明确服务的职责边界，避免循环依赖，并评估性能影响。 通过合理的设计和架构，可以构建出可维护、可扩展的 FastAPI 应用。
