FastAPI三层架构中复杂业务端点的数据聚合策略

# app/services/user_service.py
class UserService:
    """模拟用户服务，负责获取用户信息"""
    def get_user_info(self, user_id: int) -> dict:
        # 实际场景中会从数据库或调用其他微服务获取
        print(f"Fetching user info for ID: {user_id}")
        return {"user_id": user_id, "username": f"User_{user_id}", "email": f"user{user_id}@example.com"}

# app/services/product_service.py
class ProductService:
    """模拟产品服务，负责获取产品信息"""
    def get_product_info(self, product_id: int) -> dict:
        # 实际场景中会从数据库或调用其他微服务获取
        print(f"Fetching product info for ID: {product_id}")
        return {"product_id": product_id, "product_name": f"Product_{product_id}", "price": 100.0 * product_id}

# app/services/sale_service.py
class SaleService:
    """模拟销售服务，负责获取销售信息"""
    def get_sale_info(self, sale_id: int) -> dict:
        # 实际场景中会从数据库或调用其他微服务获取
        print(f"Fetching sale info for ID: {sale_id}")
        return {"sale_id": sale_id, "items_count": sale_id + 1, "total_amount": 250.0 * sale_id}

# app/services/transaction_service.py
from typing import Dict, Any

class TransactionService:
    """
    事务聚合服务，负责调用其他服务并聚合数据。
    它封装了获取完整事务信息的业务逻辑。
    """
    def __init__(self, user_svc: UserService, prod_svc: ProductService, sale_svc: SaleService):
        self.user_svc = user_svc
        self.prod_svc = prod_svc
        self.sale_svc = sale_svc

    async def get_transaction_details(self, transaction_id: int) -> Dict[str, Any]:
        # 实际场景中，transaction_id可能用于查询一个主事务记录，
        # 该记录包含关联的user_id, product_id, sale_id等
        # 这里为了简化，假设通过 transaction_id 可以推断出关联ID
        print(f"Aggregating transaction details for ID: {transaction_id}")
        user_id = transaction_id # 示例关联
        product_id = transaction_id # 示例关联
        sale_id = transaction_id # 示例关联

        # 异步调用各个服务，提高效率
        user_data = await self.user_svc.get_user_info(user_id)
        product_data = await self.prod_svc.get_product_info(product_id)
        sale_data = await self.sale_svc.get_sale_info(sale_id)

        # 聚合数据并构建最终的事务对象
        transaction_dto = {
            "transaction_id": transaction_id,
            "status": "completed",
            "timestamp": "2023-10-27T10:00:00Z",
            "user_info": user_data,
            "product_info": product_data,
            "sale_details": sale_data,
            "final_total": sale_data["total_amount"] # 示例聚合逻辑
        }
        return transaction_dto

# app/api/dependencies.py (FastAPI依赖注入配置)
from fastapi import Depends

# 依赖函数，用于提供服务实例
def get_user_service() -> UserService:
    return UserService()

def get_product_service() -> ProductService:
    return ProductService()

def get_sale_service() -> SaleService:
    return SaleService()

def get_transaction_service(
    user_svc: UserService = Depends(get_user_service),
    prod_svc: ProductService = Depends(get_product_service),
    sale_svc: SaleService = Depends(get_sale_service)
) -> TransactionService:
    """
    依赖注入TransactionService，其内部依赖其他基础服务。
    FastAPI会自动解析并提供这些依赖。
    """
    return TransactionService(user_svc, prod_svc, sale_svc)

# app/api/endpoints.py (FastAPI路由定义)
from fastapi import APIRouter
from app.api.dependencies import get_transaction_service
from app.services.transaction_service import TransactionService

router = APIRouter(prefix="/api/v1")

@router.get("/transactions/{transaction_id}", summary="获取交易详情")
async def get_transaction_endpoint(
    transaction_id: int,
    transaction_svc: TransactionService = Depends(get_transaction_service)
) -> Dict[str, Any]:
    """
    API端点，通过调用TransactionService获取聚合后的交易详情。
    """
    return await transaction_svc.get_transaction_details(transaction_id)

# main.py (FastAPI应用入口)
from fastapi import FastAPI
from app.api.endpoints import router as api_router

app = FastAPI(title="Transaction Aggregation API")
app.include_router(api_router)

# 运行应用: uvicorn main:app --reload