MagicStack / MagicStack/asyncpg

error in pytest sqlalchemy.exc.InterfaceError

Open
#1,220 1 comment 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Python
Stars
8.1k
Forks
468
PR merge metrics
No merged PRs in 30d

Description

FAILED tests/test_users.py::test_get_all_users - sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) : cannot perform operation: another ope...
FAILED tests/test_users.py::test_get_user_by_id - sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) : cannot perform operation: another ope...

db.py
```
from typing import AsyncGenerator

from sqlalchemy import NullPool
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from config import settings

engine = create_async_engine(settings.db.url)
async_session_maker = async_sessionmaker(
bind=engine, class_=AsyncSession, expire_on_commit=False
)

async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session

test_engine = create_async_engine(settings.db.test_url, poolclass=NullPool)
test_async_session_maker = async_sessionmaker(
bind=test_engine, class_=AsyncSession, expire_on_commit=False
)

async def get_test_async_session() -> AsyncGenerator[AsyncSession, None]:
async with test_async_session_maker() as session:
yield session
```

conftest.py
```
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app import app
from core.database import get_async_session, get_test_async_session, test_engine
from core.database.models import Base

@pytest_asyncio.fixture(scope="function")
async def init_db():
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)

@pytest_asyncio.fixture(scope="function")
async def session(init_db) -> AsyncSession:
async for session in get_test_async_session():
try:
yield session
except IntegrityError:
await session.rollback()
raise
finally:
await session.rollback()
await session.close()

@pytest_asyncio.fixture(scope="function")
async def client(init_db) -> AsyncClient:
async def override_get_async_session():
async for session in get_test_async_session():
yield session

app.dependency_overrides[get_async_session] = override_get_async_session

transport = ASGITransport(app=app)
client = AsyncClient(transport=transport, base_url="http://testserver/api/v1")
yield client
await client.aclose()

async def user_authentication_headers(client: AsyncClient, username: str, password: str):
json = {"username": username, "password": password}
response = await client.post("/users/login", json=json)
data = response.json()
auth_token = data["access_token"]
headers = {"Authorization": f"Bearer {auth_token}"}
return headers
```

test_users.py
```
@pytest.mark.asyncio
async def test_get_all_users(client: AsyncClient, session: AsyncSession):
UserFactory._meta.sqlalchemy_session = session
superuser = UserFactory(is_superuser=True)
session.add(superuser)
await session.commit()

headers = await user_authentication_headers(client, superuser.username, "password123")
response = await client.get(
"/users",
headers=headers
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)

@pytest.mark.asyncio
async def test_get_user_by_id(client: AsyncClient, session: AsyncSession):
UserFactory._meta.sqlalchemy_session = session
user = UserFactory(username="singleuser", is_active=True)
session.add(user)
await session.commit()

headers = await user_authentication_headers(client, user.username, "password123")
response = await client.get(
f"/users/{user.id}",
headers=headers
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "singleuser"
```

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start by running tests/test_users.py and compare the async session setup in db.py with the fixtures and dependency override in conftest.py. Focus on the failing test_get_all_users and test_get_user_by_id cases. Done means both tests complete without sqlalchemy.exc.InterfaceError.

Written by the indexing model from the issue text.

Assessment

Tech stack
postgresql, python, sqlalchemy
Domain
databases
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.