How to properly handle many to many in async sqlalchemy?

user_16350540 0
Viewed 3458

I was trying to implement many to many relationship between tables. When I use backpopulates all tags for a specific user must be in the tags field.

The tables are successfully created.

Users and tags are added. Link table too.

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.util import await_only, greenlet_spawn

from sqlalchemy import Column, Table, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.dialects.postgresql import VARCHAR, INTEGER

Base = declarative_base()

user_tag = Table('user_tag', Base.metadata,
                 Column('user_id', INTEGER, ForeignKey('users.id')),
                 Column('tag_id', INTEGER, ForeignKey('tags.id'))
                 )


class User(Base):
    __tablename__ = 'users'
    id = Column(INTEGER, primary_key=True)
    name = Column(VARCHAR(32), nullable=False, unique=True)
    tags = relationship("Tag",
                        secondary=user_tag,
                        back_populates="users")


class Tag(Base):
    __tablename__ = 'tags'
    id = Column(INTEGER, primary_key=True)
    tag = Column(VARCHAR(255), nullable=False, unique=True)
    users = relationship("User",
                         secondary=user_tag,
                         back_populates="tags")


async def main():
    engine = create_async_engine(
        "postgresql+asyncpg://postgres:pgs12345@localhost/test",
        echo=False,
    )

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

    users = [User(name="p1"), User(name="p2"), User(name="p3")]
    tags = [Tag(tag="tag1"), Tag(tag="tag2"), Tag(tag="tag3")]

    async with AsyncSession(engine) as session:
        async with session.begin():
            session.add_all(users)
            session.add_all(tags)

        for user in users:
            await session.refresh(user)
        for tag in tags:
            await session.refresh(tag)

        for user in users:
            for i in range(3, user.id - 1, -1):
                await session.execute(user_tag.insert().values(user_id=user.id, tag_id=i))
        await session.commit()

        for user in users:
            await session.refresh(user)
        for tag in tags:
            await session.refresh(tag)

        tags = await greenlet_spawn(users[0].tags)
        print(tags)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

When I run the program, it crashes with:

 File "C:\Sources\asyncSQLAl test\main.py", line 48, in <module>
    loop.run_until_complete(main())
  File "C:\Users\Stanislav\AppData\Local\Programs\Python\Python39\lib\asyncio\base_events.py", line 
642, in run_until_complete
    return future.result()
  File "C:\Sources\asyncSQLAl test\main.py", line 41, in main
    tags = await greenlet_spawn(await users[0].tags)
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\orm\attributes.py", line 480, in __get__
    return self.impl.get(state, dict_)
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\orm\attributes.py", line 931, in get
    value = self.callable_(state, passive)
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\orm\strategies.py", line 879, in _load_for_state
    return self._emit_lazyload(
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\orm\strategies.py", line 1036, 
in _emit_lazyload
    result = session.execute(
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\orm\session.py", line 1689, in 
execute
    result = conn._execute_20(statement, params or {}, execution_options)
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\engine\base.py", line 1582, in 
_execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\sql\lambdas.py", line 481, in _execute_on_connection
    return connection._execute_clauseelement(
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\engine\base.py", line 1451, in 
_execute_clauseelement
    ret = self._execute_context(
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\engine\base.py", line 1813, in 
_execute_context
    self._handle_dbapi_exception(
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\engine\base.py", line 1998, in 
_handle_dbapi_exception
    util.raise_(exc_info[1], with_traceback=exc_info[2])
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\util\compat.py", line 207, in raise_
    raise exception
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\engine\base.py", line 1770, in 
_execute_context
    self.dialect.do_execute(
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\engine\default.py", line 717, in do_execute
    cursor.execute(statement, parameters)
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py", line 449, in execute
    self._adapt_connection.await_(
  File "C:\Sources\asyncSQLAl test\venv\lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 60, in await_only
    raise exc.MissingGreenlet(
sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_() here. Was IO attempted in an unexpected place? (Background on this error at: http://sqlalche.me/e/14/xd2s)      
sys:1: RuntimeWarning: coroutine 'AsyncAdapt_asyncpg_cursor._prepare_and_execute' was never awaited

I don't quite understand how greenlet_spawn works here and where it should be used in this example.

For example, same program, but in sync style

from sqlalchemy import Column, Table, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.dialects.postgresql import VARCHAR, INTEGER

Base = declarative_base()

user_tag = Table('user_tag', Base.metadata,
                 Column('user_id', INTEGER, ForeignKey('users.id')),
                 Column('tag_id', INTEGER, ForeignKey('tags.id'))
                 )


class User(Base):
    __tablename__ = 'users'
    id = Column(INTEGER, primary_key=True)
    name = Column(VARCHAR(32), nullable=False, unique=True)
    tags = relationship("Tag",
                        secondary=user_tag,
                        back_populates="users")


class Tag(Base):
    __tablename__ = 'tags'
    id = Column(INTEGER, primary_key=True)
    tag = Column(VARCHAR(255), nullable=False, unique=True)
    users = relationship("User",
                         secondary=user_tag,
                         back_populates="tags")
    
    def __str__(self):
        return self.tag


def main():
    engine = create_engine(
        "postgresql+psycopg2://postgres:pgs12345@localhost/test",
        echo=False,
    )

    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    Session = sessionmaker(bind=engine)
    session = Session()

    users = [User(name="p1"), User(name="p2"), User(name="p3")]
    tags = [Tag(tag="tag1"), Tag(tag="tag2"), Tag(tag="tag3")]

    with session.begin():
        session.add_all(users)
        session.add_all(tags)

    for user in users:
        for i in range(3, user.id - 1, -1):
            session.execute(user_tag.insert().values(
                user_id=user.id, tag_id=i))
    session.commit()

    for tag in users[0].tags:
        print(tag, end=" ")

main()

Gives me:

tag1 tag2 tag3
python-asyncio many-to-many sqlalchemy python
2 Answers

I've been stuck on this today too and I've narrowed it down to the fact that a lazyload is attempted, which GreenLet is not happy about. I wasn't sure whether this was just my lack of skill but I've found this article that details some of the common errors: https://matt.sh/sqlalchemy-the-async-ening, where it's mentioned that this very issue will occur in this way. Furthermore, the docs go into detail about needing to avoid lazyloading: https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html.

My solution at the moment is to effectively prefetch the Child relation upon the initial query of the Parent object and then manipulate it from there. Whether this is a true bug, in the sense that it should work in async when it already works in sync or simply a limitation of the async method, I've no idea.

Edit 06/08/21, here is how I am prefetching relationships:

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from . import models

async def get_parent_prefetch_children(db: AsyncSession, parent_id: int) -> models.Parent:
    result = await db.execute(
        sa.select(models.Parent).where(models.Parent.id == parent_id).options(
            selectinload(models.Parent.children)
        )
    )
    return result.scalar()

In your case, you call users[0].tags, which makes a lazyload and fails. In order for you to avoid this, you must refetch users with their tags eagerly loaded.

user_149982130

From their official doc for relationship, you can use lazy="joined" to ask the SQLAlchemy do prefetch for you on session.refresh(...) or session.execute(select(...)). This will do joined style eagerly load at the time you query for the object. Note that this may introduce performance issue to your application since sometime you are not going to access the foreign'd objects while the database will still do joined style query for you.

tags = relationship("Tag",
    secondary=user_tag,
    back_populates="users",
    lazy="joined")

The results and SQL stmt looks like:

>>>tags = users[0].tags
>>>print(tags)
[<__main__.Tag object at 0x10bc68610>, <__main__.Tag object at 0x10bc41d00>, <__main__.Tag object at 0x10bc12b80>]
>>>from sqlalchemy.future import select
>>>print(select(User))
FROM users LEFT OUTER JOIN (user_tag AS user_tag_1 JOIN tags AS tags_1 ON tags_1.id = user_tag_1.tag_id) ON users.id = user_tag_1.user_id
user_55328760
Related