FastAPI+SQLModel外键返回关联数据抛出greenlet错误

问题现象 #

在FastAPI + SQLModel异步技术栈中，接口配置包含外键关联字段的响应模型时，抛出sqlalchemy.exc.MissingGreenlet错误，提示greenlet_spawn has not been called; can't call await_() here. Was IO attempted in an unexpected place?，即便将SQLAlchemy降级至1.4.35版本仍会复现。 #

问题相关代码 #

models.py #

class BaseQuestion(SQLModel):
    survey_id: Optional[int] = Field(default=None, foreign_key="survey.id")
    title: Optional[str] = Field(default=None, max_length=45)
    help: Optional[str] = Field(default=None, max_length=45)
    priority: Optional[int] = Field(default=None)
    is_mandatory: Optional[int] = Field(default=None)


class Question(BaseQuestion, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

    answers: List["Answer"] = Relationship(back_populates="question")
    survey: Optional[Survey] = Relationship(back_populates="questions")

class QuestionOut(BaseQuestion):
    id: Optional[int] = Field(default=None, primary_key=True)


class BaseAnswer(SQLModel):
    question_id: Optional[int] = Field(default=None, foreign_key="question.id")
    text: Optional[str] = Field(default=None, max_length=45)
    help: Optional[str] = Field(default=None, max_length=45)
    priority: Optional[str] = Field(default=None, max_length=45)
    is_stoping: Optional[int] = Field(default=0)


class Answer(BaseAnswer, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

    question: Optional[Question] = Relationship(back_populates="answers")


class AnswerOut(BaseAnswer):
    id: Optional[int] = Field(default=None, primary_key=True)


class QuestionWithAnswer(QuestionOut):
    answers: List[Answer] = []

api.py #

@app.get("/questions/{question_id}", response_model=questionnaire.QuestionOut)
async def get_question(question_id: int, session: db.Session = Depends(session)):
    db_question = await session.get(questionnaire.Question, question_id)
    if not db_question:
        raise HTTPException(status_code=404, detail="Question not found")

    return db_question

db.py #

def create_engine(host: str, username: str, password: str, name: str) -> AsyncEngine:
    return create_async_engine(
        URL.create(
            drivername="mariadb+aiomysql",
            host=host,
            username=username,
            password=password,
            database=name,
        ),
    )


def session_factory(app: FastAPI) -> Callable:
    async def session() -> AsyncGenerator[Session, None]:
        async with Session(app.state.engine) as session:
            yield session

    return session

报错信息 #

concurrency_py3k.py", line 67, in await_only
survey-service-1  |     raise exc.MissingGreenlet(
survey-service-1  | sqlalchemy.exc.StatementError: (sqlalchemy.exc.MissingGreenlet) greenlet_spawn has not been called; can't call await_() here. Was IO attempted in an unexpected place?
survey-service-1  | [SQL: SELECT answer.question_id AS answer_question_id, answer.text AS answer_text, answer.help AS answer_help, answer.priority AS answer_priority, answer.is_stoping AS answer_is_stoping, answer.id AS answer_id 
survey-service-1  | FROM answer 
survey-service-1  | WHERE %s = answer.question_id]
survey-service-1  | [parameters: [{'%(140173137451136 param)s': 1}]]

问题根因 #

• 异步模式下SQLModel/SQLAlchemy默认使用懒加载关联数据：当响应序列化阶段访问关联字段时，ORM会尝试发起隐式SQL查询，但此时没有可承载异步IO的greenlet上下文，查询无法执行await操作直接触发报错。
• 自定义的QuestionWithAnswer模型直接引用了ORM表类Answer而非纯输出类AnswerOut，序列化时会自动触发ORM关系属性的懒加载逻辑。

修复方案 #

1. 查询阶段主动预加载关联数据：禁止依赖懒加载加载关联关系，使用selectinload在查询阶段同步查出关联数据，避免序列化时发起隐式查询。
2. 响应模型禁止直接引用ORM表类：关联字段统一使用对应的输出模型，切断序列化逻辑和ORM懒加载的触发链路。
3. 会话初始化时添加expire_on_commit=False参数，避免事务提交后对象属性过期触发额外隐式查询。

修正后代码 #

修正models.py响应模型 #

# 关联字段使用输出模型AnswerOut，不要直接引用表模型Answer
class QuestionWithAnswer(QuestionOut):
    answers: List[AnswerOut] = []

修正api.py查询逻辑 #

from sqlmodel import select
from sqlalchemy.orm import selectinload

# 若需要返回关联答案，将response_model改为QuestionWithAnswer
@app.get("/questions/{question_id}", response_model=questionnaire.QuestionWithAnswer)
async def get_question(question_id: int, session: db.Session = Depends(session)):
    # 查询时主动预加载answers关联数据
    query = select(questionnaire.Question)\
        .where(questionnaire.Question.id == question_id)\
        .options(selectinload(questionnaire.Question.answers))
    result = await session.execute(query)
    db_question = result.scalar_one_or_none()
    if not db_question:
        raise HTTPException(status_code=404, detail="Question not found")
    return db_question

若不需要返回关联字段，保持response_model=QuestionOut即可，该模型未定义answers字段，序列化时不会访问关联属性，不会触发懒加载报错。

修正db.py会话配置 #

def session_factory(app: FastAPI) -> Callable:
    async def session() -> AsyncGenerator[Session, None]:
        # 添加expire_on_commit=False避免属性过期
        async with Session(app.state.engine, expire_on_commit=False) as session:
            yield session

    return session

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