alexted
commented
1 year ago I recently started getting that exception, too!
InterfaceError: cannot use Connection.transaction() in a manually started transaction
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 714, in _start_transaction
await self._transaction.start()
File "asyncpg/transaction.py", line 101, in start
raise apg_errors.InterfaceError(
AsyncAdapt_asyncpg_dbapi.InterfaceError: <class 'asyncpg.exceptions._base.InterfaceError'>: cannot use Connection.transaction() in a manually started transaction
File "sqlalchemy/engine/base.py", line 3361, in _wrap_pool_connect
return fn()
File "sqlalchemy/pool/base.py", line 325, in connect
return _ConnectionFairy._checkout(self)
File "sqlalchemy/pool/base.py", line 921, in _checkout
result = pool._dialect.do_ping(fairy.dbapi_connection)
File "sqlalchemy/engine/default.py", line 703, in do_ping
cursor.execute(self._dialect_specific_select_one)
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 479, in execute
self._adapt_connection.await_(
File "sqlalchemy/util/_concurrency_py3k.py", line 68, in await_only
return current.driver.switch(awaitable)
File "sqlalchemy/util/_concurrency_py3k.py", line 121, in greenlet_spawn
value = await result
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 408, in _prepare_and_execute
await adapt_connection._start_transaction()
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 716, in _start_transaction
self._handle_exception(error)
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 682, in _handle_exception
raise translated_error from error
InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) <class 'asyncpg.exceptions._base.InterfaceError'>: cannot use Connection.transaction() in a manually started transaction
(Background on this error at: https://sqlalche.me/e/14/rvf5)
File "starlette/middleware/exceptions.py", line 79, in __call__
raise exc
File "starlette/middleware/exceptions.py", line 68, in __call__
await self.app(scope, receive, sender)
File "fastapi/middleware/asyncexitstack.py", line 21, in __call__
raise e
File "fastapi/middleware/asyncexitstack.py", line 18, in __call__
await self.app(scope, receive, send)
File "starlette/routing.py", line 706, in __call__
await route.handle(scope, receive, send)
File "starlette/routing.py", line 276, in handle
await self.app(scope, receive, send)
File "starlette/routing.py", line 66, in app
response = await func(request)
File "fastapi/routing.py", line 225, in app
solved_result = await solve_dependencies(
File "fastapi/dependencies/utils.py", line 533, in solve_dependencies
solved = await call(**sub_values)
File "core/utils/middlewares/auth.py", line 57, in __call__
user: UserDTO | None = await self._get_user_from_db(decoded_access_token["sub"])
File "core/utils/middlewares/auth.py", line 97, in _get_user_from_db
user: UserDTO | None = await self.users_repo.get_user(UserQueryFilters(uuid=user_uuid))
File "core/data/postgresql/repositories/users.py", line 72, in get_user
user: User = await self.session.scalar(stmt)
File "sqlalchemy/ext/asyncio/session.py", line 240, in scalar
result = await self.execute(
File "sqlalchemy/ext/asyncio/session.py", line 214, in execute
result = await greenlet_spawn(
File "sqlalchemy/util/_concurrency_py3k.py", line 126, in greenlet_spawn
result = context.throw(*sys.exc_info())
File "sqlalchemy/orm/session.py", line 1713, in execute
conn = self._connection_for_bind(bind)
File "sqlalchemy/orm/session.py", line 1552, in _connection_for_bind
return self._transaction._connection_for_bind(
File "sqlalchemy/orm/session.py", line 747, in _connection_for_bind
conn = bind.connect()
File "sqlalchemy/future/engine.py", line 406, in connect
return super(Engine, self).connect()
File "sqlalchemy/engine/base.py", line 3315, in connect
return self._connection_cls(self, close_with_result=close_with_result)
File "sqlalchemy/engine/base.py", line 96, in __init__
else engine.raw_connection()
File "sqlalchemy/engine/base.py", line 3394, in raw_connection
return self._wrap_pool_connect(self.pool.connect, _connection)
File "sqlalchemy/engine/base.py", line 3364, in _wrap_pool_connect
Connection._handle_dbapi_exception_noconnection(
File "sqlalchemy/engine/base.py", line 2198, in _handle_dbapi_exception_noconnection
util.raise_(
File "sqlalchemy/util/compat.py", line 210, in raise_
raise exception
File "sqlalchemy/engine/base.py", line 3361, in _wrap_pool_connect
return fn()
File "sqlalchemy/pool/base.py", line 325, in connect
return _ConnectionFairy._checkout(self)
File "sqlalchemy/pool/base.py", line 921, in _checkout
result = pool._dialect.do_ping(fairy.dbapi_connection)
File "sqlalchemy/engine/default.py", line 703, in do_ping
cursor.execute(self._dialect_specific_select_one)
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 479, in execute
self._adapt_connection.await_(
File "sqlalchemy/util/_concurrency_py3k.py", line 68, in await_only
return current.driver.switch(awaitable)
File "sqlalchemy/util/_concurrency_py3k.py", line 121, in greenlet_spawn
value = await result
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 408, in _prepare_and_execute
await adapt_connection._start_transaction()
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 716, in _start_transaction
self._handle_exception(error)
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 682, in _handle_exception
raise translated_error from errorIn addition I periodically get two more errors: The first:
InterfaceError: cannot perform operation: another operation is in progress
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 739, in commit
self.await_(self._transaction.commit())
File "sqlalchemy/util/_concurrency_py3k.py", line 68, in await_only
return current.driver.switch(awaitable)
File "sqlalchemy/util/_concurrency_py3k.py", line 121, in greenlet_spawn
value = await result
File "asyncpg/transaction.py", line 211, in commit
await self.__commit()
File "asyncpg/transaction.py", line 179, in __commit
await self._connection.execute(query)
File "asyncpg/connection.py", line 317, in execute
return await self._protocol.query(query, timeout)
File "asyncpg/protocol/protocol.pyx", line 323, in query
self._check_state()
File "asyncpg/protocol/protocol.pyx", line 707, in asyncpg.protocol.protocol.BaseProtocol._check_state
raise apg_exc.InterfaceError(
AsyncAdapt_asyncpg_dbapi.InterfaceError: <class 'asyncpg.exceptions._base.InterfaceError'>: cannot perform operation: another operation is in progress
File "sqlalchemy/engine/base.py", line 1089, in _commit_impl
self.engine.dialect.do_commit(self.connection)
File "sqlalchemy/engine/default.py", line 686, in do_commit
dbapi_connection.commit()
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 741, in commit
self._handle_exception(error)
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 682, in _handle_exception
raise translated_error from error
InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) <class 'asyncpg.exceptions._base.InterfaceError'>: cannot perform operation: another operation is in progress
(Background on this error at: https://sqlalche.me/e/14/rvf5)
File "starlette/middleware/exceptions.py", line 79, in __call__
raise exc
File "starlette/middleware/exceptions.py", line 68, in __call__
await self.app(scope, receive, sender)
File "fastapi/middleware/asyncexitstack.py", line 15, in __call__
async with AsyncExitStack() as stack:
File "contextlib.py", line 714, in __aexit__
raise exc_details[1]
File "contextlib.py", line 697, in __aexit__
cb_suppress = await cb(*exc_details)
File "contextlib.py", line 206, in __aexit__
await anext(self.gen)
File "core/data/postgresql/engine.py", line 24, in get_db
async with async_session() as session, session.begin():
File "sqlalchemy/ext/asyncio/session.py", line 713, in __aexit__
await greenlet_spawn(
File "sqlalchemy/util/_concurrency_py3k.py", line 126, in greenlet_spawn
result = context.throw(*sys.exc_info())
File "sqlalchemy/engine/util.py", line 235, in __exit__
with util.safe_reraise():
File "sqlalchemy/util/langhelpers.py", line 70, in __exit__
compat.raise_(
File "sqlalchemy/util/compat.py", line 210, in raise_
raise exception
File "sqlalchemy/engine/util.py", line 233, in __exit__
self.commit()
File "sqlalchemy/orm/session.py", line 836, in commit
trans.commit()
File "sqlalchemy/engine/base.py", line 2459, in commit
self._do_commit()
File "sqlalchemy/engine/base.py", line 2649, in _do_commit
self._connection_commit_impl()
File "sqlalchemy/engine/base.py", line 2620, in _connection_commit_impl
self.connection._commit_impl()
File "sqlalchemy/engine/base.py", line 1091, in _commit_impl
self._handle_dbapi_exception(e, None, None, None, None)
File "sqlalchemy/engine/base.py", line 2124, in _handle_dbapi_exception
util.raise_(
File "sqlalchemy/util/compat.py", line 210, in raise_
raise exception
File "sqlalchemy/engine/base.py", line 1089, in _commit_impl
self.engine.dialect.do_commit(self.connection)
File "sqlalchemy/engine/default.py", line 686, in do_commit
dbapi_connection.commit()
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 741, in commit
self._handle_exception(error)
File "sqlalchemy/dialects/postgresql/asyncpg.py", line 682, in _handle_exception
raise translated_error from errorThe second:
AttributeError: 'NoneType' object has no attribute 'twophase'
File "starlette/middleware/exceptions.py", line 79, in __call__
raise exc
File "starlette/middleware/exceptions.py", line 68, in __call__
await self.app(scope, receive, sender)
File "fastapi/middleware/asyncexitstack.py", line 21, in __call__
raise e
File "fastapi/middleware/asyncexitstack.py", line 18, in __call__
await self.app(scope, receive, send)
File "starlette/routing.py", line 706, in __call__
await route.handle(scope, receive, send)
File "starlette/routing.py", line 276, in handle
await self.app(scope, receive, send)
File "starlette/routing.py", line 66, in app
response = await func(request)
File "fastapi/routing.py", line 225, in app
solved_result = await solve_dependencies(
File "fastapi/dependencies/utils.py", line 533, in solve_dependencies
solved = await call(**sub_values)
File "core/utils/middlewares/auth.py", line 57, in __call__
user: UserDTO | None = await self._get_user_from_db(decoded_access_token["sub"])
File "core/utils/middlewares/auth.py", line 97, in _get_user_from_db
user: UserDTO | None = await self.users_repo.get_user(UserQueryFilters(uuid=user_uuid))
File "core/data/postgresql/repositories/users.py", line 65, in get_user
result: User = await self.session.scalar(stmt)
File "sqlalchemy/ext/asyncio/session.py", line 240, in scalar
result = await self.execute(
File "sqlalchemy/ext/asyncio/session.py", line 214, in execute
result = await greenlet_spawn(
File "sqlalchemy/util/_concurrency_py3k.py", line 128, in greenlet_spawn
result = context.switch(value)
File "sqlalchemy/orm/session.py", line 1713, in execute
conn = self._connection_for_bind(bind)
File "sqlalchemy/orm/session.py", line 1552, in _connection_for_bind
return self._transaction._connection_for_bind(
File "sqlalchemy/orm/session.py", line 754, in _connection_for_bind
if self.session.twophase and self._parent is None:I don't know if there is a connection between them, but they first appeared at the same time. Сan anyone tell me what I'm doing wrong?
asyncpg: 0.27.0 PostgreSQL: 11.4 Do you use a PostgreSQL SaaS? If so, which? : Nope Can you reproduce the issue with a local PostgreSQL install?: Nope. It's very difficult because errors occur only under certain conditions under high load. Python: 3.10.8 Platform: Linux Do you use pgbouncer?: Yes, 1.10.0 Did you install asyncpg with pip?: Nope, via poetry If you built asyncpg locally, which version of Cython did you use?: - Can the issue be reproduced under both asyncio and uvloop?: I don't know. My service is based on FastAPI and runs under the hood on AnyIO. The app runs through Uvicorn, which uses uvloop under the hood.
Here's how I determine the connection to the database:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from src.app.config import config
engine = create_async_engine(
config.POSTGRES_DSN,
future=True,
echo=True,
pool_size=config.POSTGRES_MAX_CONNECTIONS,
max_overflow=config.POSTGRES_CONNECTIONS_OVERFLOW,
pool_pre_ping=True,
pool_recycle=300,
connect_args={"server_settings": {"application_name": f"{config.APP_NAME.lower()}[{config.ENVIRONMENT.lower()}]"}},
)
async_session = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
async def get_db() -> AsyncSession:
async with async_session() as session, session.begin():
yield sessionhere's my Users repository (i use repository pattern in my data layer) (Note that here I pass AsyncSession using fastapi's Depends feature):
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from .engine import get_db
from .models import User
from .dto_models import UserDTO, UserQueryFilters
class UsersRepo:
"""Repository class for work with user entities in PostgreSQL"""
def __init__(self, session: AsyncSession):
self.session: AsyncSession = session
def convert_to_dto(self, obj: User, schema):
return schema.from_orm(obj) if obj else None
async def get_user(
self,
filters: UserQueryFilters,
) -> UserDTO | None:
stmt = select(User)
schema = UserDTO
user: User = await self.session.scalar(stmt)
return self.convert_to_dto(user, schema) if user else None
def get_users_repo(db: AsyncSession = Depends(get_db)) -> UsersRepo:
return UsersRepo(db)
here's the middlware dealing with authorization (Note that here I pass UsersRepo using fastapi's Depends feature):
...
class Authenticate:
async def __call__(
self,
encoded_access_token: str = Security(oauth2_scheme),
idp_client: IDPService = Depends(get_idp_client),
users_repo: UsersRepo = Depends(get_users_repo),
):
self.idp_client = idp_client
self.users_repo = users_repo
decoded_access_token: t.Dict[str, t.Any] = self._validate_access_token(encoded_access_token)
groups, permissions = await self._get_user_roles_permissions(encoded_access_token)
user: UserDTO | None = await self._get_user_from_db(decoded_access_token["sub"])
if not user:
raise AccessDeniedError("User account was not found.")
return UserIdentity(**user.dict(), groups=groups, permissions=permissions)
@staticmethod
def _validate_access_token(access_token: str):
...
async def _get_user_roles_permissions(self, access_token: str):
...
async def _get_user_from_db(self, user_uuid) -> UserDTO | None:
"""Get user from db by uuid"""
user: UserDTO | None = await self.users_repo.get_user(UserQueryFilters(uuid=user_uuid))
return user
authenticate = Authenticate()
here's how I use this middleware in the requests handler (Note that here I pass authenticate using fastapi's Depends feature):
...
from middlewares.auth import UserIdentity, authenticate
routes: APIRouter = APIRouter()
@routes.get(
"/devices",
response_model=List[GetDevicesResp],
summary="Get devices",
description="Gives devices",
)
async def get_devices(
ids: Set[NonNegativeInt] = Query(None, alias="id", description="devices IDs", example=[1, 12, 23]),
name: str = Query(None, description="device name", example="Tablet"),
use_case: GetDevicesUseCase = Depends(get_devices_usecase),
user: UserIdentity = Depends(authenticate),
) -> List[GetDevicesResp]:
req_obj: GetDevicesReq = GetDevicesReq(user=user, ids=ids, name=name)
result: List[GetDevicesResp] = await use_case.execute(req_obj)
return resultbesides that, I also pass the db session into the use_case instance (note, in the example above, besides authenticate I also use Depends for get_devices_usecase):
import typing as t
from logging import getLogger, Logger
from fastapi import Depends
from pydantic import NonNegativeInt, NoneStr
from sqlalchemy.ext.asyncio import AsyncSession
from src.app.config import config
from src.data.postgresql.engine import get_db
from src.data.postgresql.repositories.devices import DevicesRepo
from src.data.postgresql.repositories.devices_dto import DevicesFilters, DevicesDTO
from src.use_cases.base import BaseUseCase
from src.utils.helpers.serialization_additions import OurBaseModel
from src.utils.middlewares.auth import UserIdentity
logger: Logger = getLogger(config.APP_NAME)
class GetDevicesReq(OurBaseModel):
"""Object of input data"""
ids: t.Set[int] | None
name: NoneStr
user: UserIdentity
class GetDevicesResp(OurBaseModel):
"""Object of output data"""
id: NonNegativeInt # noqa A003
name: str
class GetDevicesUseCase(BaseUseCase):
def __init__(self, devices_repo: DevicesRepo):
self.devices_repo: DevicesRepo = devices_repo
async def execute(self, req: GetDevicesReq) -> t.List[GetDevicesResp]:
"""
Implements the business logic of getting devices
:param req: object with input data
:return: devices
"""
filters: DevicesFilters = DevicesFilters(ids=req.ids, name=req.name)
devices: t.List[DevicesDTO] = await self.devices_repo.get_devices(filters)
return [GetDevicesResp(**_.dict()) for _ in devices]
async def get_devices_usecase(db: AsyncSession = Depends(get_db)) -> GetDevicesUseCase:
"""Input Boundary for GetDevicesUseCase"""
return GetDevicesUseCase(devices_repo=DevicesRepo(db))here's get_devices_usecase in turn gets the db session, also using fastapi's Depends for the previously mentioned session generator get_db
elprans
zzzeek
adriangb
kglee0
bogdanp05
unkindypie
I'm seeing
sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) <class 'asyncpg.exceptions._base.InterfaceError'>: cannot use Connection.transaction() in a manually started transactionerror, but with some inconsistency.I have a backend server running(built with FastAPI), and one of the endpoint does the following:
It is after the INSERT statement that i'm getting the error. But the funny thing is, about 1/5 times I don't get that error and the endpoint finishes successfully. I searched about the error but found little success. I dont' have code that 'explicitly' start any transactions. By looking at the logs, I do see 'implicit' begin statements(created by
asyncpgI assume?). Any idea how I could fix this error? First I thought maybe me not committing after the SELECT could be the issue, but now I'm completely lost after some of them completed successfully.