构建基于 FastAPI 的三层架构：多服务协同处理复杂端点-Python教程-PHP中文网

构建基于 FastAPI 的三层架构：多服务协同处理复杂端点

聖光之護

发布： 2025-08-28 18:05:01

原创

797人浏览过

构建基于 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 对象，然后将该对象返回给应用层。

即构数智人

即构数智人是由即构科技推出的AI虚拟数字人视频创作平台，支持数字人形象定制、短视频创作、数字人直播等。

查看详情

示例代码：

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) 或聚合服务。

总结：

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

以上就是构建基于 FastAPI 的三层架构：多服务协同处理复杂端点的详细内容，更多请关注php中文网其它相关文章！