FastAPI三层架构中复杂端点多服务协作与聚合策略

# app/schemas/transaction.py
from pydantic import BaseModel

class UserInfo(BaseModel):
    id: str
    name: str
    email: str

class ProductInfo(BaseModel):
    id: str
    name: str
    price: float

class SaleDetails(BaseModel):
    order_id: str
    quantity: int
    total_amount: float

class TransactionDTO(BaseModel):
    id: str
    user_info: UserInfo
    product_info: ProductInfo
    sale_details: SaleDetails

# app/services/user_service.py (示例，实际可能通过HTTP请求或ORM)
class UserService:
    async def get_user_by_transaction(self, transaction_id: str) -> UserInfo:
        # 模拟从用户服务获取数据
        print(f"Fetching user data for transaction {transaction_id}")
        return UserInfo(id="user123", name="John Doe", email="john@example.com")

# app/services/product_service.py
class ProductService:
    async def get_product_by_transaction(self, transaction_id: str) -> ProductInfo:
        # 模拟从产品服务获取数据
        print(f"Fetching product data for transaction {transaction_id}")
        return ProductInfo(id="prod456", name="Laptop Pro", price=1200.00)

# app/services/sale_service.py
class SaleService:
    async def get_sale_details(self, transaction_id: str) -> SaleDetails:
        # 模拟从销售服务获取数据
        print(f"Fetching sale data for transaction {transaction_id}")
        return SaleDetails(order_id="ord789", quantity=1, total_amount=1200.00)

# app/api/endpoints.py
from fastapi import APIRouter, Depends
from app.services.user_service import UserService
from app.services.product_service import ProductService
from app.services.sale_service import SaleService
from app.schemas.transaction import TransactionDTO, UserInfo, ProductInfo, SaleDetails

router = APIRouter()

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

def get_product_service():
    return ProductService()

def get_sale_service():
    return SaleService()

@router.get("/transactions/{transaction_id}", response_model=TransactionDTO)
async def get_transaction_option1(
    transaction_id: str,
    user_service: UserService = Depends(get_user_service),
    product_service: ProductService = Depends(get_product_service),
    sale_service: SaleService = Depends(get_sale_service)
) -> TransactionDTO:
    """
    策略一：应用层直接协调多服务
    """
    print("Executing Option 1: Application layer orchestration")
    # 应用层直接调用并聚合数据
    user_data = await user_service.get_user_by_transaction(transaction_id)
    product_data = await product_service.get_product_by_transaction(transaction_id)
    sale_data = await sale_service.get_sale_details(transaction_id)

    # 组合数据构建TransactionDTO
    transaction_dto = TransactionDTO(
        id=transaction_id,
        user_info=user_data,
        product_info=product_data,
        sale_details=sale_data
    )
    return transaction_dto

# app/services/transaction_service.py
from app.services.user_service import UserService
from app.services.product_service import ProductService
from app.services.sale_service import SaleService
from app.schemas.transaction import TransactionDTO, UserInfo, ProductInfo, SaleDetails

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_details(self, transaction_id: str) -> TransactionDTO:
        """
        聚合逻辑封装在TransactionService中
        """
        print(f"TransactionService: Aggregating data for transaction {transaction_id}")
        user_data = await self.user_service.get_user_by_transaction(transaction_id)
        product_data = await self.product_service.get_product_by_transaction(transaction_id)
        sale_data = await self.sale_service.get_sale_details(transaction_id)

        transaction_dto = TransactionDTO(
            id=transaction_id,
            user_info=user_data,
            product_info=product_data,
            sale_details=sale_data
        )
        return transaction_dto

# app/api/endpoints.py (续)
from fastapi import APIRouter, Depends
# 假设已经定义了UserService, ProductService, SaleService和TransactionDTO
from app.services.transaction_service import TransactionService

# 依赖注入函数
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)

@router.get("/transactions_v2/{transaction_id}", response_model=TransactionDTO)
async def get_transaction_option2(
    transaction_id: str,
    transaction_service: TransactionService = Depends(get_transaction_service)
) -> TransactionDTO:
    """
    策略二：应用层调用专用聚合服务
    """
    print("Executing Option 2: Application layer calls aggregate service")
    # 应用层只负责调用聚合服务
    return await transaction_service.get_transaction_details(transaction_id)