aiopg + pgbouncer

[vineetgoel]

I am currently using asyncpg and looks like it doesn't play well with pgbouncer. How does aiopg play with pgbouncer?

[vir-mir]

yes, we use together with a pg_bouncer. works

[decaz]

@vir-mir @asvetlov I use aiopg.sa and I want to use PgBouncer as a local pool on the same host instead of embedded pool, - how can I disable pooling on the aiopg side? Within SQLAlchemy it can be done by using sqlalchemy.pool.NullPool.

[vir-mir]

@decaz Hi! I think we will make it possible to pass user pool_cls parameters in the future.

In this situation, you can use the connection and nested context managers. Example:

import aiopg

async def test_one_connect(pg_params):
    async with await aiopg.connect(**pg_params) as conn, conn.cursor() as cur:
        assert not conn.closed
        assert not cur.closed

        await cur.execute('DROP TABLE IF EXISTS sa_tbl5')
        await cur.execute(
            'CREATE TABLE sa_tbl5 ('
            '"id" VARCHAR(255) NOT NULL, '
            '"name" VARCHAR(255), '
            'PRIMARY KEY ("id"))'
        )
        assert not conn.closed
        assert not cur.closed

    assert conn.closed
    assert cur.closed

    async with await aiopg.connect(**pg_params) as conn, conn.cursor() as cur:
        assert not conn.closed
        assert not cur.closed

        await cur.execute(
            "INSERT INTO sa_tbl5 (id, name) VALUES ('id-test', 'name-test')"
        )

        await cur.execute("SELECT * FROM sa_tbl5")
        row = await cur.fetchone()
        assert set(row) == set(['id-test', 'name-test'])

        assert not conn.closed
        assert not cur.closed

    assert conn.closed
    assert cur.closed

Or you can create a fake pool, for to hide nested context managers and pass settings db once Example:

import aiopg

class ContextManagerConnect:
    def __init__(self, conn_coro):
        self._conn_coro = conn_coro
        self._conn = None
        self._cur = None

    async def __aenter__(self):
        self._conn = await self._conn_coro
        self._cur = await self._conn.cursor()
        return self._cur

    async def __aexit__(self, exc_type, exc, tb):
        self._cur.close()
        self._conn.close()


class FakePool:
    def __init__(self, **kwargs):
        self._pg_params = kwargs

    def connect(self):
        return ContextManagerConnect(aiopg.connect(**self._pg_params.copy()))

async def test_fake_pool(pg_params):
    null_pool = FakePool(**pg_params)

    async with null_pool.connect() as conn:
        assert not conn.closed
        assert not conn.connection.closed

        await conn.execute('DROP TABLE IF EXISTS sa_tbl5')

        assert not conn.closed
        assert not conn.connection.closed

        await conn.execute(
            'CREATE TABLE sa_tbl5 ('
            '"id" VARCHAR(255) NOT NULL, '
            '"name" VARCHAR(255), '
            'PRIMARY KEY ("id"))'
        )
        assert not conn.closed
        assert not conn.connection.closed

    assert conn.closed
    assert conn.connection.closed

    async with null_pool.connect() as conn:
        assert not conn.closed
        assert not conn.connection.closed

        await conn.execute(
            "INSERT INTO sa_tbl5 (id, name) VALUES ('id-test', 'name-test')"
        )
        assert not conn.closed
        assert not conn.connection.closed

        await conn.execute("SELECT * FROM sa_tbl5")
        row = await conn.fetchall()
        assert set(row[0]) == set(['id-test', 'name-test'])

        assert not conn.closed
        assert not conn.connection.closed

    assert conn.closed
    assert conn.connection.closed