Обработка «InterfaceError: выполняется другая операция» с помощью Async SQLAlchemy и FastAPI ⇐ Python

[Anonymous]

I'm developing a FastAPI application where I use SQLModel with the asyncpg driver for asynchronous database operations. Despite following the asynchronous patterns and ensuring proper await usage on database calls, I encounter the following error during my pytest tests:

InterfaceError: cannot perform operation: another operation is in progress

This error arises when executing database operations, seemingly due to concurrent access or overlapping database transactions. I've tried ensuring that each test and request uses its own AsyncSession and that all sessions and transactions are properly closed and committed.

import random import string import pytest import pytest_asyncio from httpx import AsyncClient, ASGITransport from main import app # Make sure this import points to your FastAPI app instance @pytest_asyncio.fixture async def client(): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: yield client @pytest_asyncio.fixture def generate_random_phone_number(): def _generate(length=10): return ''.join(random.choices(string.digits, k=length)) return _generate @pytest_asyncio.fixture def generate_random_phone_prefix(): def _generate(): prefix_length = random.randint(1, 3) return '+' + ''.join(random.choices(string.digits, k=prefix_length)) return _generate @pytest.mark.asyncio async def test_create_user(client: AsyncClient, generate_random_phone_number, generate_random_phone_prefix): user_data = { "phone_number": generate_random_phone_number(), "phone_prefix": generate_random_phone_prefix() } response = await client.post("/api/user/", json=user_data) assert response.status_code == 201 data = response.json() assert data["phone_number"] == user_data["phone_number"] assert data["phone_prefix"] == user_data["phone_prefix"] @pytest.mark.asyncio async def test_duplicate_user(client: AsyncClient, generate_random_phone_number, generate_random_phone_prefix): phone_number = generate_random_phone_number() phone_prefix = generate_random_phone_prefix() user_data = { "phone_number": phone_number, "phone_prefix": phone_prefix } await client.post("/api/user/", json=user_data) response = await client.post("/api/user/", json=user_data) assert response.status_code == 400 data = response.json() assert data["detail"] == "A user with the given phone number and prefix already exists." @pytest.mark.asyncio async def test_create_interest(client: AsyncClient, generate_random_phone_number, generate_random_phone_prefix): # First, create a user user_data = { "phone_number": generate_random_phone_number(), "phone_prefix": generate_random_phone_prefix() } user_response = await client.post("/api/user/", json=user_data) assert user_response.status_code == 201 user = user_response.json() interest_data = { "topic": "Sample Topic", "found": 1, "search": True } headers = {"User-ID": str(user["id"])} response = await client.post("/api/interest/", json=interest_data, headers=headers) assert response.status_code == 201 interest = response.json() assert interest["topic"] == interest_data["topic"] assert interest["found"] == interest_data["found"] assert interest["search"] == interest_data["search"] @pytest.mark.asyncio async def test_get_interest(client: AsyncClient, generate_random_phone_number, generate_random_phone_prefix): user_data = { "phone_number": generate_random_phone_number(), "phone_prefix": generate_random_phone_prefix() } user_response = await client.post("/api/user/", json=user_data) assert user_response.status_code == 201 user = user_response.json() headers = {"User-ID": str(user["id"])} interest_1 = { "topic": "Sample Topic1", "found": 1, "search": True } response = await client.post("/api/interest/", json=interest_1, headers=headers) assert response.status_code == 201 interest_2 = { "topic": "Sample Topic2", "found": 0, "search": True } response = await client.post("/api/interest/", json=interest_2, headers=headers) assert response.status_code == 201 response = await client.get("/api/interest/", headers=headers) interests = response.json() assert len(interests) == 2 # Validate the content of the first interest object interest_1_response = interests[0] assert interest_1_response["topic"] == "Sample Topic1" assert interest_1_response["found"] == 1 assert interest_1_response["search"] is True assert "created_at" in interest_1_response assert "updated_at" in interest_1_response assert interest_1_response["created_at"]

Источник: https://stackoverflow.com/questions/780 ... -sqlalchem