pg-force-execute 0.0.6

Context manager to run PostgreSQL queries with SQLAlchemy, terminating any other clients that block them

pg-force-execute

Context manager to run PostgreSQL queries with SQLAlchemy, terminating any other clients that continue to block it after a configurable delay.

Using this to wrap queries is somewhat of a last resort, but is useful in certain Extract Transform Load (ETL) pipeline contexts. For example, if it is more important to replace one table with another than to allow running queries on the table to complete, then this can be used to run the relevant ALTER TABLE RENAME TO query.

Installation

pip install pg-force-execute

Example usage

import datetime
import sqlalchemy as sa
from pg_force_execute import pg_force_execute

# Run postgresql locally should allow the below to run
# docker run --rm -it -e POSTGRES_HOST_AUTH_METHOD=trust -p 5432:5432 postgres

engine = sa.create_engine('postgresql://postgres@127.0.0.1:5432/')
query = 'SELECT 1'  # A more realistic example would be something that needs an exclusive lock on a table

with \
        engine.begin() as conn, \
        pg_force_execute(
            conn,           # SQLAlchemy connection to run the query
            engine,         # SQLAlchemy engine that will create new connections to cancel blocking queries
            delay=datetime.timedelta(minutes=5),  # Amount of time to wait before cancelling queries
        ):

    results = conn.execute(query)
    print(results.fetchall())

API

The API a single context manager pg_force_execute.

pg_force_execute(conn, engine, delay=datetime.timedelta(minutes=5), check_interval=datetime.timedelta(seconds=1), termination_thread_timeout=datetime.timedelta(seconds=10), logger=logging.getLogger("pg_force_execute"))

• conn - A SQLAlchemy connection that will be unblocked

• engine - A SQLAlchemy engine to create a new connection that will be used to terminate backends blocking conn

• delay (optional) - How long to wait before attempting to terminate backends blocking conn

• check_interval (optional) - The interval between repeated attempted to terminate backends blocking conn

• termination_thread_timeout (optional) - How long to wait for the termination to complete

• logger (optional) The Python logger instance through which to log

Running tests locally

pip install -e ".[dev]"  # Only needed once
./start-services.sh      # Only needed once
pytest