[Python] SQLAlchemy, FastAPI, Dishka. Cannot use Connection.transaction() in a manually...

The code is used in many projects and everything works. But in one service, sometimes an error occurs on the prod:

sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) <class 'asyncpg.exceptions._base.InterfaceError'>: cannot use Connection.transaction() in a manually started transaction


Moreover, everything is OK on Dev and locally.

The problem is that I can't figure out what is being bumped into on the prod that leads to such an error, and therefore I have no idea how to solve the problem, because everything is OK in the rest of the services and in this service on dev.

I am sure that there is no manual transaction management anywhere in the code except in the IoC container and the consumer.

Here is the IoC code of the container:

from dishka import Provider, Scope, provide
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy.orm import sessionmaker

from config import Config, get_config
from infrastructure.db.sqlalchemy.setup import create_engine, create_session_pool
from infrastructure.uow.sqlalchemy import SQLAlchemyUnitOfWork


class BaseProvider(Provider):
@provide(scope=Scope.APP)
def config(self) -> Config:
return get_config()

@provide(scope=Scope.APP)
def sqlalchemy_engine(self, config: Config) -> AsyncEngine:
return create_engine(config.pg_config)

@provide(scope=Scope.APP)
def session_pool(self, sqlalchemy_engine: AsyncEngine) -> sessionmaker:
return create_session_pool(sqlalchemy_engine)

@provide(scope=Scope.REQUEST)
def sqlalchemy_uow(self, session_factory: sessionmaker) -> SQLAlchemyUnitOfWork:
return SQLAlchemyUnitOfWork(_session=session_factory())


setup.py

from typing import Callable, AsyncContextManager

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker

from config import PGConfig
from infrastructure.db.sqlalchemy.models import BaseModel


def create_engine(db: PGConfig, echo: bool = False) -> AsyncEngine:
engine = create_async_engine(
db.pg_database_url,
echo=echo
)
return engine


async def create_tables(engine: AsyncEngine) -> None:
async with engine.begin() as conn:
await conn.run_sync(BaseModel.metadata.create_all)


def create_session_pool(engine: AsyncEngine) -> Callable[[], AsyncContextManager[AsyncSession]]:
session_pool = sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
return session_pool


And here is the UoW code (In it I tried to solve this problem by removing the await self line._session.begin(), but it didn't help):

from dataclasses import dataclass
from types import TracebackType
from typing import Self

from sqlalchemy.ext.asyncio import AsyncSession

from infrastructure.repositories.invite_links.sqlalchemy import SQLAlchemyInviteLinksRepository
from infrastructure.repositories.tags.sqlalchemy import SQLAlchemyTagsRepository, SQLAlchemyTagsTypesRepository
from infrastructure.uow.base import BaseUnitOfWork


@dataclass
class SQLAlchemyUnitOfWork(BaseUnitOfWork):
_session: AsyncSession = None

async def __aenter__(self) -> Self:
# await self._session.begin()
return self

async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None
) -> None:
try:
if exc_type is None:
await self._session.commit()
else:
await self._session.rollback()
finally:
await self._session.close()

@property
def invite_links(self) -> SQLAlchemyInviteLinksRepository:
self._session_check()
return SQLAlchemyInviteLinksRepository(self._session)

@property
def tags(self) -> SQLAlchemyTagsRepository:
self._session_check()
return SQLAlchemyTagsRepository(self._session)

@property
def tags_types(self) -> SQLAlchemyTagsTypesRepository:
self._session_check()
return SQLAlchemyTagsTypesRepository(self._session)

def _session_check(self) -> None:
if not self._session:
raise ValueError("Session is not initialized")


UoW is used in Fastap points (everything is done according to Dishka documentation).

And I also throw the session factory into the consumer.:

lifespan.py:

from contextlib import asynccontextmanager

import aiojobs
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy.orm import sessionmaker

from app.api.background import consume_in_background
from config import Config
from infrastructure.db.sqlalchemy.setup import create_tables
from infrastructure.message_brokers.kafka import KafkaMessageBroker


@asynccontextmanager
async def lifespan(app: FastAPI) -> FastAPI:
container = app.state.dishka_container
engine = await container.get(AsyncEngine)
scheduler = aiojobs.Scheduler()
broker = await container.get(KafkaMessageBroker)
config = await container.get(Config)
if config.app.is_dev or config.app.is_production:
session_factory = await container.get(sessionmaker)
job = await scheduler.spawn(consume_in_background(broker, session_factory, config))
await create_tables(engine)
yield
if config.app.is_dev or config.app.is_production:
await job.close()
await scheduler.close()
await container.close()


background.py

import logging

from sqlalchemy.orm import sessionmaker

from config import Config
from domain.entities.invite_links import InviteLink
from infrastructure.message_brokers.base import BaseMessageBroker
from infrastructure.repositories.filters.invite_links import InviteLinksFilter
from infrastructure.uow.sqlalchemy import SQLAlchemyUnitOfWork


async def consume_in_background(
broker: BaseMessageBroker,
session_factory: sessionmaker,
config: Config,
) -> None:
logger = logging.getLogger("LinkConsumer")
async for message_dict in broker.start_consuming():
message = message_dict["value"]
topic = message_dict["topic"]
link = message.get("invite_link")
channel_id = message.get("channel_id")
try:
if topic == config.rp_config.CONSUME_TOPIC:
if not link:
continue
link = InviteLink(
channel_id=channel_id,
**link
)
uow = SQLAlchemyUnitOfWork(_session=session_factory())
async with uow as unit_of_work:
if not await unit_of_work.invite_links.is_exists(link.id, link.channel_id):
await unit_of_work.invite_links.create_link(link)
elif topic == config.rp_config.CONSUME_MERGE_ACCOUNTS_TOPIC:
uow = SQLAlchemyUnitOfWork(_session=session_factory())
async with uow as unit_of_work:
links = await unit_of_work.invite_links.get_all_links(
links_filter=InviteLinksFilter(
page_size=-1,
creator_id=message["old_account_id"],
)
)
links = links[0]
for link in links:
link.creator_id = message["new_account_id"]
await unit_of_work.invite_links.update_link(link)
await broker.commit()
except Exception as e:
logger.error(f"Error while consuming message: {e}", exc_info=True)


I made echo = True and monitored the number of transactions opening when using the application, noticed that sometimes unnecessary ones appeared and decided to comment on the "await self line._session.begin()", there seem to be fewer transactions, but the problem returned after a few days on the prod.

Right now I'm just restarting the service to temporarily solve it.

Continue reading...