Support for asynchronous cursors

[princess-entrapta]

Since version 3 psycopg supports asynchronous cursors as part of the core lib. As I am writing a modern async server, I ran into the issue this module only supports synchronous cursors from postgres.

It's a significant problem since I need to test my codebase who heavily awaits cursors, results etc. and would of course ensure this code is correct before running it into production.

Fortunately, you might be interested to know this support can be added using very minimal effort. I leave you to determine the exact way to implement the option in your lib, but my approach was very simply to duplicate the postgresql factory in my own code and change a few lines here and there.

def async_postgresql(
    # [...] Skipping doc and params
    @pytest.fixture
    # This must return an async fixture now
    async def postgresql_factory(request: FixtureRequest) -> Iterator[connection]:
        """
        Async fixture factory for PostgreSQL.

        :param request: fixture request object
        :returns: postgresql client
        """
       # [...] Skipping body

        with DatabaseJanitor(
            pg_user, pg_host, pg_port, pg_db, proc_fixture.version, pg_password, isolation_level
        ) as janitor:
            # Line modified here
            db_connection: connection = await psycopg.AsyncConnection.connect(
                dbname=pg_db,
                user=pg_user,
                password=pg_password,
                host=pg_host,
                port=pg_port,
                options=pg_options,
            )
            for load_element in pg_load:
                janitor.load(load_element)
            yield db_connection
            # And here
            await db_connection.close()

    return postgresql_factory

Hope this helps. Available for any question.

[fizyk]

@arthur-hav how do you run your test? Also async? Where do you use those cursors? in the testing or tested code?

[rodrigoalmeida94]

@arthur-hav I would second your suggestion! Implementing an async cursor for this library would be very useful.

@fizyk in order to run async tests I often use the pytest-asyncio library, e.g.:

@pytest.mark.asyncio
async def test_some_asyncio_code():
    res = await library.do_something()
    assert b"expected result" == res

[princess-entrapta]

@arthur-hav how do you run your test? Also async? Where do you use those cursors? in the testing or tested code?

Everything is async, both tests and product code. We run tests using anyio iirc in a similar manner than Rodrigo mentioned

[fizyk]

Hmm.... Then I guess It should be okay. I was afraid you wanted to feed the client to the testing code instead of relying only to check the database.

If you'll provide both client and tests that tests the client, that would be great :)

[princess-entrapta]

I'm not sure what do you mean by client, I imagine that's the code that I test who is typically client of postgres, in any case that's wrong please correct me. As I understand you want to see my general usage of the factory I derived from your factory and why/how it is useful so I'll detail that

I use a FastApi app that is asynchronous so my entrypoint is an async function that calls different services that are also async. I don't run any asyncio loop myself, the loop is run by the webserver itself. Tests use pytest.mark.anyio for the same purpose.

My server code is bound with the repository through dependency injection. Here is a rough simplification of how this works:

# -- web app --

async def postgres_handler():
    return await PostgresRepository.connect()

@router.get("/")
async def get_something(handler: MyHandler = Depends(postgres_handler)) -> Iterable[Something]:
    return await handler.get_something()

# -- database repository --
from psycopg import AsyncConnection
from psycopg.rows import class_row

class PostgresRepository:
    @classmethod
    async def connect(cls):
        conn = await AsyncConnection.connect(
            host=os.environ.get("DB_HOST", ""),
            user=os.environ.get("DB_LOGIN", ""),
            password=os.environ.get("DB_PASSWORD", ""),
        )
        return cls(conn)

    def __init__(self, connection) -> None:
        self.conn = connection

    def get_cursor(self, *args, **kwargs):
        return self.conn.cursor(*args, **kwargs)

    async def get_something():
        cur = self.get_cursor(row_factory=class_row(Something))
        await cur.execute("SELECT * FROM some_table;")
        return await cur.fetchall()

The associated tests:

# -- fixtures --

## factory I described at issue start + some omitted fixture for the web app

@pytest.fixture
async def postgres_fixture(async_postgresql):
    async def fake_conn():
        return PostgresRepository(async_postgresq)

    with mock.patch.object(PostgresRepository, "connect", fake_conn):
        conn = await fake_conn()
        cur = conn.get_cursor()
        await cur.execute(open("tests/test.sql").read())  ## Loads test data
        yield cur

# -- test --

pytestmark = pytest.mark.anyio

async def test_get_something(postgres_fixture, client):
    response = await client.get("/")
    assert response.status_code == 200
    assert len(response.json()) == 2 ## The number of entries in test.sql for some_table

[fizyk]

sorry, my bad, not client but connection fixture factory. So.... implementation and tests.

[fizyk]

Okay should be solved in a manner of examples rather than actual implementation, see #894

[txross1993]

Note for the above implementation at least on python 3.11, pytest==8.3.3, pytest-asyncio==0.24.0, you will need to annotate your fixtures with @pytest_asyncio.fixture instead of @pytest.fixture, otherwise they return an async generator.