FastAPI数据库上下文

Question

我想我误解了依赖注入是如何在FastAPI中使用的，特别是在DB会话的上下文中

我目前的设置是FastAPI、SqlAlchhemy&；Alembic，虽然我自己在编写原始SQL，但pydantic等，非常直截了当

我有一些基本的CRUD路由，可以直接与我的存储库层通信，而且一切都正常。在这些方法中，我能够成功地使用DB依赖项注入。请参见下面的示例代码：

依赖关系

def get_database(request: Request) -> Database:
    return request.app.state._db

def get_repository(Repo_type: Type[BaseRepository]) -> Callable:
    def get_repo(db: Database = Depends(get_database)) -> Type[BaseRepository]:
        return Repo_type(db)
    return get_repo

通过ID路线获取的示例

@router.get("/{id}/", response_model=TablePub, name="Get Table by id")
async def get_table_by_id(
    id: UUID, table_repo: TableRepository = Depends(get_repository(TableRepository))
) -> TableInDB:
    table = await table_repo.get_table_by_id(id=id)
    if not table:
        raise HTTPException(status_code=HTTP_404_NOT_FOUND, detail="No Table found with that id.")
    return table

对应的存储库

from databases import Database

class BaseRepository:
    def __init__(self, db: Database) -> None:
        self.db = db

class TableRepository(BaseRepository):

    async def get_table_by_id(self, *, id: UUID) -> TableInDB:
        table = await self.db.fetch_one(
            query=GET_TABLE_BY_ID_QUERY, 
            values={"id": id},
        )
        if not table:
            return None
        return TableInDB(**table)

现在我想开始做一些更复杂的操作，并想添加一个服务层来容纳所有的业务逻辑

什么是正确的结构方式，以便我可以重用我已经编写的存储库？例如，我想返回一个表的所有销售额，但我需要先从数据库中获取表号，然后才能查询销售表。路由要求以参数形式传入表\u id->；服务层，其中我按ID获取表（使用现有repo）->；从该对象中，获取表号，然后向需要表号作为参数的外部API发出请求

到目前为止，我所拥有的：

路线

@router.get("/{table_id}", response_model=SalesPub, name="Get Sale Entries by table id")
async def get_sales_by_table_id(
    table_id: UUID = Path(..., title="ID of the Table to get Sales Entries for")):
    response = await SalesService.get_sales_from_external_API(table_id=table_id)
    return response

服务层“SalesService”

async def get_sales_from_external_API(
        table_id: UUID, 
        table_repo: TableRepository = Depends(get_repository(TableRepository))
        ) -> TableInDB:
        table_data = await table_repo.get_table_by_id(id=table_id)
        if table_data is None:
            logger.info(f"No table with id:{table_id} could not be found")
        table_number = table_data.number
        client_id = table_data.client_id
        
        sales = await salesGateway.call_external_API(table_number, client_id)
        return sales

代码在此停止table_data = await table_repo.get_table_by_id(id=table_id)

有一个错误AttributeError: 'Depends' object has no attribute 'get_table_by_id'

我不明白的是，代码几乎与可以通过ID获取表的route方法相同？Dependes对象TableRepository确实有一个get_table_by_id方法。我做得不对的是什么？这是将业务逻辑与数据库操作分离的最佳方法吗

提前谢谢

Answer 1

我似乎找到了解决这个问题的办法，尽管我不确定这是否是最好的办法

Depends模块仅适用于FastAPI路由和依赖项。我试图在常规函数中使用它

我需要使参数table_repo成为Depends的实例。并将其作为参数传递给外部API调用函数

@router.get("/table/{table_id}/", response_model=SalePub, name="Get Sales by table id")
    async def get_sales_by_table_id(
        table_id: UUID = Path(..., title="ID of the Table to get Sales Entries for"),
        table_repo: TableRepository = Depends(get_repository(TableRepository))):
        response = await get_sales_entries_from_pos(table_id=table_id, table_repo=table_repo)
        return response

我预见的问题是，如果我有一个可能需要访问manny repos的大型服务，我必须通过Dependes在路由器上进行访问，这对我来说有点奇怪