Link Tables getting duplicate entries only under async?

[jidn]

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

"""Sample to show issues between Session and AsyncSession.

# Setup environment

uv init sqlmodel-link-table
cd sqlmodel-link-table
uv venv
source .venv/bin/activate
uv pip install sqlmodel aiosqlite greelet pytest pytest-asyncio


Copy this file to sqlmodel-link-table/test_session.py and run pytest.
    $ pytest test_session.py -v
"""

from contextlib import asynccontextmanager, contextmanager
from typing import AsyncGenerator, Generator
from uuid import UUID, uuid4

import pytest
from sqlalchemy import UniqueConstraint
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine


class LinkOrgUser(SQLModel, table=True):
    """Roles a user has in an Organization."""

    __tablename__: str = "organization_user"  # type: ignore

    org_id: UUID | None = Field(
        default=None, foreign_key="organization.id", primary_key=True
    )
    user_id: UUID | None = Field(default=None, foreign_key="user.id", primary_key=True)
    role: int  # enum.IntFlag

    org: "Organization" = Relationship(back_populates="users")
    user: "User" = Relationship(back_populates="orgs")


class User(SQLModel, table=True):
    """User with relationships."""

    __table_args__ = (UniqueConstraint("email"),)

    id: UUID = Field(primary_key=True, default_factory=uuid4)
    name: str = Field(max_length=64)
    email: str | None = Field(default=None, max_length=255)
    # Relationships
    orgs: list[LinkOrgUser] = Relationship(
        back_populates="user",
    )


class Organization(SQLModel, table=True):
    """Organization with users."""

    id: UUID = Field(primary_key=True, default_factory=uuid4)
    name: str = Field(max_length=80)
    users: list[LinkOrgUser] = Relationship(back_populates="org")


###########################################################
##  pytest with synchronous database session             ##


@contextmanager
def memory_session() -> Generator[Session]:
    """Syncronious database session."""
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False})

    SQLModel.metadata.create_all(bind=engine)
    with Session(engine) as session:
        yield session


def test_link_sync():
    with memory_session() as session:
        org = Organization(name="Example", id=UUID(int=1))
        aa = User(name="AA", id=UUID(int=2))
        org.users.append(LinkOrgUser(org=org, user=aa, role=1))

        session.add(org)
        session.commit()

        assert org.name == "Example"
        assert 1 == len(org.users)
        assert aa.id in [_.user.id for _ in org.users]


###########################################################
##  pytest with asynchronous database session            ##


@asynccontextmanager
async def async_memory_session() -> AsyncGenerator[AsyncSession]:
    """Async database session."""
    url = "sqlite+aiosqlite:///:memory:"
    engine = create_async_engine(url, echo=False, future=True)
    async with async_sessionmaker(engine, expire_on_commit=False)() as session:
        async with engine.begin() as conn:
            await conn.run_sync(SQLModel.metadata.create_all)
        try:
            yield session
        finally:
            print(f"{url} finished.")
    await engine.dispose()


@pytest.mark.asyncio
async def test_link_async():
    async with async_memory_session() as async_session:
        org = Organization(name="Example", id=UUID(int=5))
        aa = User(name="AA", id=UUID(int=6))
        org.users.append(LinkOrgUser(org=org, user=aa, role=1))

        async_session.add(org)
        await async_session.commit()

        assert org.name == "Example"
        assert 1 == len(org.users)
        assert aa.id in [_.user.id for _ in org.users]

Description

I create an Organization and User objects. I then create a link table object using the two objects and the user's role. I commit and check the number of users in the Organization.users object.

I expect to see one user, which I do when using a synchronous database session. However under an asynchronous database session, I see two duplicates.

assert 1 == len(org.users)
E AssertionError: assert 1 == 2
E + where 2 = len([LinkOrgUser(role=1, >org_id=UUID('...005'), user_id=UUID('...006')), LinkOrgUser(role=1, >org_id=UUID('...005'), user_id=UUID('...006'))])
E + where [LinkOrgUser(role=1, org_id=UUID('...005'), user_id=UUID('...006')), LinkOrgUser(role=1, org_id=UUID('...005'), user_id=UUID('...006'))] = Organization(id=UUID('...005'), name='Example').users

There are no errors other than the assert in the failed test. I don't understand why one session works while the other doesn't. I must be doing something wrong, but what?

Operating System

Linux

Operating System Details

Up to date Arch

SQLModel Version

0.0.22

Python Version

3.13.0

Additional Context

Package Version

---

aiosqlite 0.20.0
annotated-types 0.7.0
greenlet 3.1.1
iniconfig 2.0.0
packaging 24.2
pluggy 1.5.0
pydantic 2.9.2
pydantic-core 2.23.4
pytest 8.3.3
pytest-asyncio 0.24.0
sqlalchemy 2.0.36
sqlmodel 0.0.22
typing-extensions 4.12.2