FastAPI Strawberry GraphQL中如何正确使用get_db函数？

你的错误根源是get_db()是一个带yield的生成器函数，直接调用它返回的是生成器对象，而不是SQLAlchemy的Session实例，自然没有query属性。以下是两种解决方案：

方案1：手动获取会话（临时应急） #

通过next()迭代生成器，取出真正的数据库会话，同时手动触发生成器的finally块关闭会话：

@strawberry.type
class Query:
    @strawberry.field
    async def get_blog(self) -> List[ShowBlogType]:
        db_generator = get_db()
        db = next(db_generator)
        try:
            blogs = BlogRepository.get_all_blog(db)
            return blogs
        finally:
            # 触发生成器的finally逻辑，关闭数据库连接
            try:
                next(db_generator)
            except StopIteration:
                pass

这种方法仅适合临时测试，代码繁琐且不符合最佳实践。

方案2：使用Strawberry依赖注入（推荐） #

Strawberry支持FastAPI的Depends依赖注入机制，能自动管理数据库会话的生命周期，这是标准且优雅的做法：

步骤1：在字段方法中注入数据库会话 #

修改Query的字段定义，通过Depends(get_db)将Session实例注入到方法参数中：

from strawberry.fastapi import Depends
from sqlalchemy.orm import Session
from fastapi import run_in_threadpool

@strawberry.type
class Query:
    @strawberry.field
    async def get_blog(self, db: Session = Depends(get_db)) -> List[ShowBlogType]:
        # 用线程池包装同步数据库操作，避免阻塞异步事件循环
        blogs = await run_in_threadpool(BlogRepository.get_all_blog, db)
        return blogs

步骤2：确保Strawberry与FastAPI集成正确 #

如果是通过FastAPI挂载Strawberry GraphQL，需保证路由配置正确：

from strawberry.fastapi import GraphQLRouter
import strawberry

schema = strawberry.Schema(query=Query)
graphql_router = GraphQLRouter(schema)

# 将GraphQL路由挂载到FastAPI应用
app.include_router(graphql_router, prefix="/graphql")

这样Strawberry会自动处理依赖注入，每次请求生成新的数据库会话，请求结束后自动关闭连接。

额外注意事项 #

如果BlogRepository.get_all_blog是同步数据库操作，必须用run_in_threadpool包装成异步执行，否则会阻塞FastAPI的事件循环，影响服务性能。

内容的提问来源于stack exchange，提问作者Sharifu