多层架构中依赖倒置实现疑问（FastAPI场景）

你的实现存在几个关键问题，同时可以通过调整设计更贴合洋葱架构的依赖倒置原则：

现存问题 #

1. 类型提示错误：process函数的参数标注为ProductRepository，但你实际定义的抽象接口是IProductRepository，这会导致类型检查失败，违背了"依赖抽象而非具体实现"的核心原则。
2. 多重继承顺序不合理：Python的方法解析顺序（MRO）会影响抽象方法的检查逻辑，当前ProductSqlRepository(SqlRepository, IProductRepository)的顺序中，具体存储基类在前，抽象接口在后，可能导致抽象方法的实现检查不严格（比如如果SqlRepository意外覆盖抽象方法但不符合IProductRepository的要求）。
3. 抽象方法定义不规范：ABC中的抽象方法应该明确标注返回类型，用...替代pass更符合Python抽象类的规范写法。
4. 未适配FastAPI依赖注入：当前实例化仓储的方式无法直接对接FastAPI的依赖注入系统，不利于后续在接口中通过注入切换存储实现。

优化方案 #

1. 修正抽象接口与继承顺序 #

调整继承顺序，让领域抽象接口（IProductRepository）排在具体存储基类之前，确保MRO优先检查抽象方法的实现，同时规范抽象方法定义：

from abc import ABC, abstractmethod
from typing import Optional, TypeVar
from sqlalchemy import func

from pydantic import BaseModel
from pymongo.collection import Collection
from sqlalchemy.orm import Session

# 定义泛型类型，方便后续扩展
T = TypeVar('T')

class ProductDto(BaseModel):
    title: str
    weight: float

class IRepository(ABC):
    """通用仓储接口：定义基础CRUD操作"""
    @abstractmethod
    def get(self, id_) -> Optional[T]:
        ...

    @abstractmethod
    def create(self, create_data: dict) -> T:
        ...

class SqlRepository(IRepository):
    """SQLAlchemy通用仓储实现"""
    model = None  # 子类需指定对应的SQLAlchemy模型

    def __init__(self, session: Session):
        self._session = session

    def get(self, id_) -> Optional[T]:
        print(f'getting instance id={id_} in SqlRepository')
        return self._session.query(self.model).get(id_)

    def create(self, create_data: dict) -> T:
        instance = self.model(**create_data)
        self._session.add(instance)
        self._session.commit()
        self._session.refresh(instance)
        print(f'creating instance in SqlRepository')
        return instance

class MongoRepository(IRepository):
    """MongoDB通用仓储实现"""
    def __init__(self, collection: Collection):
        self._collection = collection

    def get(self, id_) -> Optional[T]:
        print(f'getting instance id={id_} in MongoRepository')
        return self._collection.find_one({"_id": id_})

    def create(self, create_data: dict) -> T:
        result = self._collection.insert_one(create_data)
        print(f'creating instance in MongoRepository')
        return self._collection.find_one({"_id": result.inserted_id})

class IProductRepository(IRepository):
    """产品领域专属仓储接口：扩展基础CRUD"""
    @abstractmethod
    def get_random_product(self) -> Optional[ProductDto]:
        ...

# 调整继承顺序：先继承领域抽象接口，再继承具体存储实现
class ProductSqlRepository(IProductRepository, SqlRepository):
    """SQL版产品仓储实现"""
    model = Product  # 假设存在对应的SQLAlchemy Product模型

    def get_random_product(self) -> Optional[ProductDto]:
        print('getting random product in ProductSqlRepository')
        instance = self._session.query(self.model).order_by(func.random()).first()
        return ProductDto.from_orm(instance) if instance else None

class ProductMongoRepository(IProductRepository, MongoRepository):
    """MongoDB版产品仓储实现"""
    def get_random_product(self) -> Optional[ProductDto]:
        print('getting random product in ProductMongoRepository')
        cursor = self._collection.aggregate([{"$sample": {"size": 1}}])
        instance = cursor.next() if cursor.alive else None
        return ProductDto(**instance) if instance else None

2. 适配FastAPI依赖注入 #

通过FastAPI的Depends实现仓储的注入，轻松切换存储实现：

from fastapi import FastAPI, Depends

app = FastAPI()

# 提供SQL会话依赖
def get_sql_session() -> Session:
    session = Session()  # 实际应使用你的SQLAlchemy会话工厂
    try:
        yield session
    finally:
        session.close()

# 提供Mongo集合依赖
def get_mongo_collection() -> Collection:
    from pymongo import MongoClient
    client = MongoClient("mongodb://localhost:27017")
    db = client["test_db"]
    yield db["products"]

# 定义仓储依赖：可通过配置动态切换存储
def get_product_repo(
    sql_session: Session = Depends(get_sql_session),
    mongo_collection: Collection = Depends(get_mongo_collection)
) -> IProductRepository:
    # 可通过环境变量等方式动态选择存储类型
    # import os
    # return ProductSqlRepository(sql_session) if os.getenv("STORAGE_TYPE") == "SQL" else ProductMongoRepository(mongo_collection)
    return ProductSqlRepository(sql_session)

@app.get("/products/{product_id}")
def get_product(product_id: int, repo: IProductRepository = Depends(get_product_repo)):
    product = repo.get(product_id)
    return product

@app.get("/products/random")
def get_random_product(repo: IProductRepository = Depends(get_product_repo)):
    product = repo.get_random_product()
    return product

3. 关键设计亮点 #

• 依赖倒置：所有业务逻辑都依赖IProductRepository抽象接口，而非具体的SQL/Mongo实现，符合洋葱架构核心思想。
• 职责划分清晰：通用仓储处理基础CRUD，领域仓储接口定义专属业务方法，具体实现类负责整合两者并实现专属逻辑。
• 灵活切换：通过修改依赖注入函数或配置变量，无需改动业务代码即可切换存储方式。

内容的提问来源于stack exchange，提问作者Альберт Александров