Блог

В настоящее время я использую асинхронную SQLAlchemy с классом, который имеет асинхронные методы и работает в asyncio цикле событий.

 from asyncio.tasks import current_task
from sqlalchemy import update as select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession

engine = create_async_engine(url)
async_session_factory = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
AsyncSession = scoped_session(async_session_factory, scopefunc=current_task)

class Animals:
    def __init__(self):
        self.connection = None

    async def connect(self):
        self.connection = AsyncSession()

    async def add():
        query = select(Animals.nicknames)
        self.connection.add(query)
        self.connection.commit()

    async def get():
        query = select(Animals.nicknames)
        rows = await self.connection.execute(query)
        for row in rows:
            print(row[0])
 

Мой класс должен иметь

 self.connection = AsyncSession()
 

потому что приложение, использующее этот класс, вызовет Animals.connection.close() завершение работы.

Время от времени будет появляться сообщение об ошибке:

sqlalchemy.exc.InvalidRequestError: Этот сеанс находится в состоянии «готов»; в рамках этой транзакции больше не может быть отправлен SQL.

а также

sqlalchemy.диалекты.postgresql.asyncpg.AsyncAdapt_asyncpg_dbapi.Ошибка интерфейса: Ошибка интерфейса’>: не удается выполнить операцию: выполняется другая операция

Можно ли избежать этой ошибки без изменения внешнего приложения, использующего этот класс?