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.