dvarrazzo

"""
PostgreSQL Logical Replication for psycopg3

A production-ready implementation of PostgreSQL logical replication using psycopg3.
Works around psycopg3's missing replication support by using a raw socket bridge.

Author: Richard Brandes
Date: October 2025
License: MIT

dvarrazzo

import statistics as st
from time import sleep, monotonic
from argparse import ArgumentParser, Namespace

import psycopg
import psycopg2
from psycopg import sql

DSN = "dbname=psycopg3_test host=localhost sslmode=disable"

dvarrazzo

#!/usr/bin/env python
"""Convert a codebase to using the walrus operator.

Hint: in order to explore the AST of a module you can run:

    python -m ast path/to/module.py

"""

from __future__ import annotations

dvarrazzo

class DictTuple(tuple):
    """Tuple class with added item getting by name.
    """
    def __new__(cls, d):
        rv = super().__new__(cls, d.values())
        rv._map = d
        return rv

    def __repr__(self):
        return f"{type(self).__qualname__}({self._map!r})"

dvarrazzo

dvarrazzo

name: Tests

on:
  push:
    branches: [ master ]
  pull_request:
    branches: [ master ]

jobs:
  tests-python:

dvarrazzo

#!/bin/bash

# Return the ip address for a container in docker-compose

set -euo pipefail
# set -x

if [[ "${1:-}" == "" ]]; then
    echo "usage: $0 container

dvarrazzo

#!/usr/bin/env python3
"""
Apply database patches.

Applied patches are recorded in the schema_patch table of the database.

The dsn to connect to defaults to a local one (empty connection string). It can
be chosen using the command line or an environment variable. Patches
application is interactive by default.

dvarrazzo

r"""Implementation of a `stan`_\ -like XML description language.

.. _stan: http://www.kieranholland.com/code/documentation/nevow-stan/

An example XHTML page can be generated by:

.. python::

    def items():
        return [ T.li['foo'], T.li['bar'], T.li['baz'] ]

dvarrazzo

"""
A quick test to try and run psycopg3 with anyio
"""

import anyio
from psycopg3 import exceptions as exc
from psycopg3.waiting import Wait, Ready
from psycopg3.connection import AsyncConnection

import logging