AsyncSession.refresh on self referencing relationships don't load correctly

[A00474880]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

class Node(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    parent_node_id: uuid.UUID | None = Field(
        default=None, foreign_key="node.id", nullable=True, ondelete="SET NULL"
    )
    parent_node: Optional["Node"] = Relationship(
        back_populates="sub_nodes",
        sa_relationship_kwargs=dict(remote_side="Node.id", lazy="raise"),
    )
    sub_nodes: list["Node"] = Relationship(
        back_populates="parent_node", sa_relationship_kwargs={"lazy": "raise"}
    )


async def create_node(session: AsyncSession) -> Node:
    node = Node(parent_node_id="19e42ad3-d0f6-48ea-a93b-bcb9f2de7718") #reference an existing node
    session.add(node)
    await session.commit()
    await session.refresh(node, attribute_names["parent_node", "sub_nodes"]
    return node

Description

In the example above, the node instance doesn't have parent_node loaded properly and the field throws:
InvalidRequestError: 'Node.parent_node' is not available due to lazy='raise'

sub_nodes loads properly and has value of empty list []

Using:
sqlalchemy Version: 2.0.44
sqlmodel Version: 0.0.27

Operating System

Linux

Operating System Details

No response

SQLModel Version

0.0.27

Python Version

3.13.5

Additional Context

I'ved opened a similar discussion in SqlAlchemy and determined that this is specific to SQLModel.
SQLAlchemy test scripts:

import asyncio
from typing import List
from typing import Optional

from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


class Node(Base):
    __tablename__ = "node"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    parent_node_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("node.id"), nullable=True
    )
    parent_node: Mapped[Optional["Node"]] = relationship(
        back_populates="sub_nodes", remote_side="Node.id", lazy="raise"
    )
    sub_nodes: Mapped[List["Node"]] = relationship(
        back_populates="parent_node", lazy="raise"
    )


async def main():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")

    try:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

        async with AsyncSession(engine) as session:
            parent = Node(id=1, name="parent")
            child = Node(id=2, name="child", parent_node_id=parent.id)
            session.add(parent)
            session.add(child)
            await session.commit()

            assert "parent_node" not in child.__dict__
            assert "sub_nodes" not in child.__dict__
            await session.refresh(child, attribute_names=["parent_node", "sub_nodes"])

            print(f"Child's parent: {child.parent_node}")
            print(f"Child's sub_nodes: {child.sub_nodes}")
    finally:
        await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

Output:

Child's parent: <__main__.Node object at 0x7b36c11c6510>
Child's sub_nodes: []

SQLModel testscript:

import asyncio
from typing import Optional
import uuid

from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import Field, Relationship, SQLModel


class Node(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    parent_node_id: uuid.UUID | None = Field(
        default=None, foreign_key="node.id", nullable=True, ondelete="SET NULL"
    )
    parent_node: Optional["Node"] = Relationship(
        back_populates="sub_nodes",
        sa_relationship_kwargs=dict(remote_side="Node.id", lazy="raise"),
    )
    sub_nodes: list["Node"] = Relationship(
        back_populates="parent_node", sa_relationship_kwargs={"lazy": "raise"}
    )


async def main():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")

    try:
        async with engine.begin() as conn:
            await conn.run_sync(SQLModel.metadata.create_all)

        async with AsyncSession(engine) as session:
            parent = Node(name="parent")
            child = Node(name="child", parent_node_id=parent.id)
            session.add(parent)
            session.add(child)
            await session.commit()

            assert "parent_node" not in child.__dict__
            assert "sub_nodes" not in child.__dict__
            await session.refresh(child, attribute_names=["parent_node", "sub_nodes"])

            print(f"Child's parent: {child.parent_node}")
            print(f"Child's sub_nodes: {child.sub_nodes}")
    finally:
        await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

Output:

Child's parent: 
Child's sub_nodes: []

[A00474880]

Here's the sqlalchemy thread for anyone interested.
sqlalchemy/sqlalchemy#13081

As discussed, in my actual use case I have set expire_on_commit=False which makes the above testcase work. I will try and add a real reproduction when I have time to demo the problem better.

[amasen02]

There are two main reasons why this happens in SQLModel with self-referential relationships, and how to configure it so both parent_node and sub_nodes refresh properly under AsyncSession with lazy='raise'.

---

1. remote_side String Resolution in SQLModel

In raw SQLAlchemy Declarative, string references like remote_side="Node.id" are resolved late against the declarative class registry. However, in SQLModel, table classes inherit from Pydantic BaseModel and compile fields into SQLAlchemy Table columns through SQLModel's custom metaclass and registry.

When remote_side="Node.id" is passed as a plain string inside sa_relationship_kwargs, SQLModel passes it directly to SQLAlchemy's relationship factory before the mapper registry finishes mapping Node. When remote_side fails to resolve the exact column target, SQLAlchemy cannot determine the foreign key direction for parent_node (whether it is Many-to-One or One-to-Many). As a result, during session.refresh(child, attribute_names=["parent_node"]), it issues a query looking in the wrong direction or skips loading the scalar parent, leaving the attribute unpopulated in the instance state. When you subsequently access child.parent_node, it triggers the lazy-load handler, which raises:

InvalidRequestError: 'Node.parent_node' is not available due to lazy='raise'

To fix this, supply remote_side as a callable lambda returning [Node.id]:

class Node(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    parent_node_id: uuid.UUID | None = Field(
        default=None, foreign_key="node.id", nullable=True, ondelete="SET NULL"
    )
    parent_node: Optional["Node"] = Relationship(
        back_populates="sub_nodes",
        sa_relationship_kwargs={
            "remote_side": lambda: [Node.id],
            "lazy": "raise",
        },
    )
    sub_nodes: list["Node"] = Relationship(
        back_populates="parent_node",
        sa_relationship_kwargs={"lazy": "raise"},
    )

Wrapping in lambda: [Node.id] ensures evaluation is deferred until after Node is fully defined and registered in SQLAlchemy's mapper.

---

2. Verified Working Script

Here is the complete, working reproduction with lambda: [Node.id], showing that session.refresh cleanly loads both parent_node and sub_nodes:

import asyncio
from typing import Optional
import uuid

from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import Field, Relationship, SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession


class Node(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    name: str = Field(default="")
    parent_node_id: uuid.UUID | None = Field(
        default=None, foreign_key="node.id", nullable=True, ondelete="SET NULL"
    )
    parent_node: Optional["Node"] = Relationship(
        back_populates="sub_nodes",
        sa_relationship_kwargs={
            "remote_side": lambda: [Node.id],
            "lazy": "raise",
        },
    )
    sub_nodes: list["Node"] = Relationship(
        back_populates="parent_node",
        sa_relationship_kwargs={"lazy": "raise"},
    )


async def main():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")

    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    async with AsyncSession(engine) as session:
        parent = Node(name="parent")
        session.add(parent)
        await session.commit()
        await session.refresh(parent)

        child = Node(name="child", parent_node_id=parent.id)
        session.add(child)
        await session.commit()

        # Refresh both relationships on the child
        await session.refresh(child, attribute_names=["parent_node", "sub_nodes"])

        print(f"Child's parent: {child.parent_node.name} (id: {child.parent_node.id})")
        print(f"Child's sub_nodes: {child.sub_nodes}")

    await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

Output:

Child's parent: parent (id: ...)
Child's sub_nodes: []

---

3. Alternative: Query-Level Eager Loading with selectinload

When working in async environments with lazy="raise", relying on session.refresh for relationships requires an extra SQL roundtrip per refreshed attribute. The standard async pattern in SQLModel/SQLAlchemy is to load relationships explicitly using selectinload:

from sqlmodel import select
from sqlalchemy.orm import selectinload

statement = (
    select(Node)
    .where(Node.id == child.id)
    .options(
        selectinload(Node.parent_node),
        selectinload(Node.sub_nodes),
    )
)
result = await session.exec(statement)
node = result.one()

This issues single, efficient queries for the related entities and avoids any risk of hitting the lazy='raise' barrier.