SQLModel中如何动态切换懒加载与急加载策略？

问题 #

使用SQLModel的Relationship()简化代码时，遇到异步驱动（如asyncpg）下懒加载关联对象会抛出"greenlet error"。当前只能在定义Relationship()时全局设置lazy="selectin"等急加载策略，但实际需求是按需加载——有时需要获取Conversation的messages，有时不需要。尝试用SQLAlchemy的.options()动态设置加载策略，但未生效。

原因 #

异步SQLAlchemy驱动中，查询结束后会话会自动关闭，此时触发懒加载会因会话已关闭报错；同步驱动则允许懒加载时复用会话，因此无此问题。

解决方案 #

SQLModel完全兼容SQLAlchemy的动态加载策略，只需正确使用加载器并在查询时添加.options()。步骤如下：

1. 重置模型的默认加载策略：将Relationship()中的lazy设为"raise"（避免意外触发懒加载报错）或保留默认的"select"（但异步场景下不要直接访问关联对象，除非会话处于活跃状态）。
2. 按需在查询时指定加载策略：使用SQLAlchemy的selectinload、joinedload等加载器，通过.options()添加到查询语句中，实现动态加载关联对象。

修改后的代码示例 #

模型定义（重置lazy策略）

from typing import Optional, Sequence
from uuid import UUID, uuid4
from sqlmodel import SQLModel, Relationship, Field, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload  # 导入加载器

class Citation(SQLModel, table=True):
    id: Optional[UUID] = Field(
        default_factory=uuid4,
        primary_key=True,
        description="The unique identifier of the citation",
    )
    content: str
    message_id: UUID | None = Field(
        foreign_key="message.id",
        ondelete="CASCADE",
        index=True,
        description="The unique identifier of the message the citation belongs to",
    )
    message: "Message" = Relationship(
        back_populates="citations", sa_relationship_kwargs={"lazy": "raise"}
    )

class Message(SQLModel, table=True):
    id: Optional[UUID] | None = Field(
        default_factory=uuid4,
        primary_key=True,
        description="The unique identifier of the message",
    )
    conversation_id: UUID | None = Field(
        foreign_key="conversation.id",
        ondelete="CASCADE",
        index=True,
        description="The unique identifier of the conversation the message belongs to",
    )
    content: str
    conversation: "Conversation" = Relationship(
        back_populates="messages", sa_relationship_kwargs={"lazy": "raise"}
    )
    citations: list["Citation"] = Relationship(
        back_populates="message",
        sa_relationship_kwargs={"lazy": "raise", "cascade": "all, delete-orphan"},
    )

class Conversation(SQLModel, table=True):
    id: Optional[UUID] | None = Field(
        default_factory=uuid4,
        primary_key=True,
        description="The unique identifier of the conversation",
    )
    title: str
    creator_id: int
    messages: list["Message"] = Relationship(
        back_populates="conversation",
        sa_relationship_kwargs={"lazy": "raise", "cascade": "all, delete-orphan"},
    )

动态加载的查询方法

# 不加载messages的查询（仅获取Conversation基本信息）
async def get_conversations(session: AsyncSession) -> Sequence[Conversation]:
    conversations = (
        await session.exec(
            select(Conversation).where(Conversation.creator_id == 1)
        )
    ).all()
    return conversations

# 加载messages的查询（同时获取Conversation和关联的messages）
async def get_conversations_with_messages(session: AsyncSession) -> Sequence[Conversation]:
    conversations = (
        await session.exec(
            select(Conversation)
            .where(Conversation.creator_id == 1)
            .options(selectinload(Conversation.messages))  # 动态添加加载策略
        )
    ).all()
    return conversations

说明 #

• selectinload会发起额外的批量查询加载关联对象，适合一对多关系，避免N+1问题；若需要联表查询，可替换为joinedload。
• 设置lazy="raise"后，若未在查询时指定加载策略，直接访问关联对象会抛出异常，避免因会话关闭导致的模糊错误。

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