It throw asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress I use an asyncpg, my code: in the test

import unittest
from models.create import create_city
from models.conn import pool_instance


class CreateTables(unittest.IsolatedAsyncioTestCase):
    async def test_create_city1(self):
        pool = await pool_instance.get_pool()
        await create_city("Moscow")
        async with pool.acquire() as conn:
            city = await conn.fetchrow("SELECT * FROM cities WHERE name = 'Moscow'")
            self.assertEqual(city["name"], "Moscow")
    
    async def test_create_city2(self):
        pool = await pool_instance.get_pool()
        await create_city("SPB")
        async with pool.acquire() as conn:
            city = await conn.fetchrow("SELECT * FROM cities WHERE name = 'SPB'")
            self.assertEqual(city["name"], "SPB")

in the models.create.create_city

async def create_city(name: str) -> None:
    pool = await pool_instance.get_pool()
    async with pool.acquire() as conn:
        await conn.execute('INSERT INTO cities(name) VALUES($1)', name)

in the models.conn

from config import DB_DSN
from asyncpg import create_pool

class ConnectionPool:
    _pool = None

    async def get_pool(self):
        if self._pool is None:
            self._pool = await create_pool(dsn=DB_DSN)
        return self._pool

pool_instance = ConnectionPool()

Is there an error in writing the tests themselves or in using the pool? How can I fix the tests or maybe the interaction with the database itself?

I tried to run these functions through asyncio.gather and it didn't throw an error, I don't understand why this happens in tests. I'm interested in how to make tests and interact with the database.

0

There are 0 best solutions below