FastAPI中基于类的视图实现CRUD遇依赖注入错误求助

首先明确：FastAPI完全支持通用CRUD路由的实现，你的思路没问题，报错原因是对Depends的使用场景和通用路由的注册逻辑理解有误。

错误原因分析 #

1. AttributeError: 'Depends' object has no attribute 'list'
   你大概率是把Depends直接绑定到了路由类的属性上（比如self.service = Depends(...)），然后直接调用self.service.list()。但Depends本身只是一个标记类，不是实际的服务实例，只有在路由函数内部被FastAPI解析后，才会生成对应的服务对象，直接访问Depends对象的方法自然会报错。

2. AttributeError: 'Depends' object has no attribute 'routes'
include_router要求传入的是APIRouter实例，而你用Depends包裹了这个参数，导致FastAPI去访问Depends对象的routes属性，这显然不符合预期。

正确实现方案 #

1. 先完善通用BaseService（基于你的已有实现） #

from sqlalchemy.orm import Session
from typing import TypeVar, Generic, Type

ModelType = TypeVar("ModelType")

class BaseService(Generic[ModelType]):
    def __init__(self, model: Type[ModelType], db: Session):
        self.model = model
        self.db = db

    def list(self, skip: int = 0, limit: int = 100):
        return self.db.query(self.model).offset(skip).limit(limit).all()

    def get(self, item_id: int):
        return self.db.query(self.model).get(item_id)

    def create(self, obj_in):
        db_obj = self.model(**obj_in.dict())
        self.db.add(db_obj)
        self.db.commit()
        self.db.refresh(db_obj)
        return db_obj

    def update(self, db_obj, obj_in):
        for key, value in obj_in.dict(exclude_unset=True).items():
            setattr(db_obj, key, value)
        self.db.commit()
        self.db.refresh(db_obj)
        return db_obj

    def delete(self, db_obj):
        self.db.delete(db_obj)
        self.db.commit()
        return db_obj

2. 实现通用BaseRouter #

核心是在路由函数内部通过Depends获取DB会话，再实例化对应服务，而不是在类属性层面绑定Depends：

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import TypeVar, Generic, Type, List
from pydantic import BaseModel

from your_db_module import get_db  # 后续会实现这个DB会话依赖

ModelType = TypeVar("ModelType")
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
ServiceType = TypeVar("ServiceType", bound=BaseService)

class BaseRouter(Generic[ModelType, CreateSchemaType, UpdateSchemaType, ServiceType]):
    def __init__(self, 
                 model: Type[ModelType],
                 service_cls: Type[ServiceType],
                 create_schema: Type[CreateSchemaType],
                 update_schema: Type[UpdateSchemaType],
                 prefix: str):
        self.model = model
        self.service_cls = service_cls
        self.router = APIRouter(prefix=prefix, tags=[model.__name__])
        self.register_crud_routes(create_schema, update_schema)

    def register_crud_routes(self, create_schema, update_schema):
        # 列表接口
        @self.router.get("/", response_model=List[ModelType])
        def get_list(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
            service = self.service_cls(self.model, db)
            return service.list(skip=skip, limit=limit)

        # 详情接口
        @self.router.get("/{item_id}", response_model=ModelType)
        def get_detail(item_id: int, db: Session = Depends(get_db)):
            service = self.service_cls(self.model, db)
            item = service.get(item_id)
            if not item:
                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
            return item

        # 创建接口
        @self.router.post("/", response_model=ModelType, status_code=status.HTTP_201_CREATED)
        def create_item(item_in: create_schema, db: Session = Depends(get_db)):
            service = self.service_cls(self.model, db)
            return service.create(obj_in=item_in)

        # 更新接口
        @self.router.put("/{item_id}", response_model=ModelType)
        def update_item(item_id: int, item_in: update_schema, db: Session = Depends(get_db)):
            service = self.service_cls(self.model, db)
            db_item = service.get(item_id)
            if not db_item:
                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
            return service.update(db_obj=db_item, obj_in=item_in)

        # 删除接口
        @self.router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
        def delete_item(item_id: int, db: Session = Depends(get_db)):
            service = self.service_cls(self.model, db)
            db_item = service.get(item_id)
            if not db_item:
                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
            service.delete(db_obj=db_item)

3. 数据库会话自动管理依赖 #

创建your_db_module.py，实现自动创建和关闭会话的依赖：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 根据你的数据库类型修改URL
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}  # SQLite专属参数，其他数据库可移除
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

4. 实际使用通用路由 #

from fastapi import FastAPI
from your_models import User  # 你的SQLAlchemy模型
from your_schemas import UserCreate, UserUpdate  # 你的Pydantic请求/响应模型

# 创建用户模块的通用路由
user_router = BaseRouter(
    model=User,
    service_cls=BaseService,
    create_schema=UserCreate,
    update_schema=UserUpdate,
    prefix="/users"
).router

# 注册到主应用
app = FastAPI()
app.include_router(user_router)

扩展说明 #

• 如果某个模型需要自定义CRUD逻辑，只需继承BaseService实现专属方法，在创建BaseRouter时传入该专属服务类即可。
• 通用路由已经集成了DB会话的自动管理，每个请求都会获取独立会话，请求结束后自动关闭，无需手动处理。

内容的提问来源于stack exchange，提问作者Che el Horn