Replication support

[dvarrazzo]

Add support for streaming replication as in psycopg2

[ghost]

Hi,

If this still an issue, I could take a look and use your guidance to implement this.

[dvarrazzo]

If you would like to contribute to it you are welcome.

[dufferzafar]

@dvarrazzo What sort of support exists today to get logical replication working in an async mode?

I guess logical replication subscription as supported by psycopg2.extras - ReplicationCursor is not currently in psycopg3.

Can I still somehow use the SQL queries directly like (CREATE SLOT, START REPLICATION) to get changes as they happen?

asyncpg mentions this snippet of code in one of their issues regarding replication: MagicStack/asyncpg#91

async def poll_replication_stream(conn):
    while True:
        async for r in conn.fetch('SELECT * FROM pg_logical_slot_get_changes(...)'):
            await put_to_queue(r)
        await asyncio.sleep(polling_interval)

...

conn = await asyncpg.connect(...)
loop.create_task(poll_replication_stream(conn))

I'm assuming something like this will work with psycopg3 as well?

[dufferzafar]

@dvarrazzo Could you please also explain exactly what would need to be added to support replication directly in psycopg3?

I'm taking a look at the code right now & having some pointers on exactly what needs to be done would be great!

[dvarrazzo]

Replication support is not implemented in psycopg 3. It needs complete redesign and implementation. If you want to contribute its implementation you are welcome, but it is a major work.

[dufferzafar]

Yeah, I realised that after looking at the code of pg2 & pg3. I guess it'll require a lot of interfacing with pq as well?

My current plan is to use something like wal2json extension on the server, then create a slot and get_changes in the JSON format directly from the server.

Or perhaps I won't do this in async mode and just use pg2 directly. I found pypgoutput which does client side decoding of pgoutput format to JSON: https://github.com/dgea005/pypgoutput