Testing PostgreSQL

Warnings

• breaking This library requires a pre-installed PostgreSQL server in your system's PATH environment variable. It uses binaries like `initdb` and `postgres` directly. Without these, the library cannot function.
  Fix: Ensure that your system has PostgreSQL installed and that its binary directory (e.g., `/usr/bin/`, `/usr/local/bin/`) is included in your PATH environment variable.
• gotcha The `testing-postgresql` project appears to be minimally maintained, with the last PyPI release in 2016 and limited GitHub activity since 2017. Users seeking active development, support for newer PostgreSQL features, or compatibility with recent Python versions might find it lacking.
  Fix: Evaluate alternatives like `pytest-postgresql` for more actively developed solutions, especially if you require modern features, community support, or encounter compatibility issues with newer Python or PostgreSQL versions.
• gotcha By default, `testing.postgresql.Postgresql` invokes `initdb` on every instantiation. This can be a slow operation, especially when creating many test databases for individual test cases, leading to increased test suite execution time.
  Fix: For faster repeated tests, use `testing.postgresql.PostgresqlFactory` with `cache_initialized_db=True`. This initializes the database only once and copies it for subsequent instances. You can also use the `copy_data_from` option to provide a pre-existing data directory.
• gotcha Failure to explicitly call `.stop()` on a `Postgresql` instance or neglecting to use it within a `with` statement (context manager) can leave lingering PostgreSQL processes and temporary data directories, consuming system resources and potentially causing conflicts.
  Fix: Always use `testing.postgresql.Postgresql()` within a `with` statement to ensure proper cleanup. If manual management is necessary (e.g., in `setUp`/`tearDown` methods of `unittest.TestCase`), ensure `postgresql_instance.stop()` is called in `tearDown`.

Install

• pip install testing.postgresql Install stable version

Imports

• Postgresql

  from testing.postgresql import Postgresql

• PostgresqlFactory

  from testing.postgresql import PostgresqlFactory

  Used for creating cached PostgreSQL instances to speed up tests.

Quickstart

This quickstart demonstrates launching a temporary PostgreSQL instance using `testing.postgresql.Postgresql` within a context manager. It shows how to connect to the instance using both SQLAlchemy and psycopg2, execute basic queries, and ensures the PostgreSQL server is properly terminated and cleaned up. It also includes an example of `PostgresqlFactory` which caches the initialized database to speed up repeated test runs, optionally with a custom initialization handler.

import testing.postgresql
from sqlalchemy import create_engine
import psycopg2

# Basic usage with a context manager
with testing.postgresql.Postgresql() as postgresql:
    # Connect using SQLAlchemy
    engine = create_engine(postgresql.url())
    with engine.connect() as conn:
        result = conn.execute(postgresql.text("SELECT 1")).scalar()
        print(f"SQLAlchemy: Connected to {postgresql.url()}, result: {result}")

    # Connect using psycopg2 (requires psycopg2 to be installed)
    try:
        conn_psycopg = psycopg2.connect(**postgresql.dsn())
        cursor = conn_psycopg.cursor()
        cursor.execute("CREATE TABLE my_table (id SERIAL PRIMARY KEY, name VARCHAR(255))")
        cursor.execute("INSERT INTO my_table (name) VALUES (%s)", ('test_name',))
        cursor.execute("SELECT name FROM my_table WHERE id = 1")
        name_result = cursor.fetchone()[0]
        print(f"psycopg2: Inserted and retrieved: {name_result}")
        cursor.close()
        conn_psycopg.close()
    except Exception as e:
        print(f"psycopg2 connection/query failed: {e}")

# Example of using PostgresqlFactory for faster tests (e.g., in a test suite setup)
def custom_init_handler(pg_instance):
    """An optional handler to run SQL on a newly initialized DB."""
    conn = psycopg2.connect(**pg_instance.dsn())
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE users (id SERIAL PRIMARY KEY, username VARCHAR(255))")
    conn.commit()
    conn.close()
    print("Factory DB initialized with 'users' table.")

PostgresqlFactory = testing.postgresql.PostgresqlFactory(
    cache_initialized_db=True, 
    on_initialized=custom_init_handler
)

with PostgresqlFactory() as postgresql_cached:
    conn = psycopg2.connect(**postgresql_cached.dsn())
    cursor = conn.cursor()
    cursor.execute("INSERT INTO users (username) VALUES (%s)", ('cached_user',))
    conn.commit()
    cursor.execute("SELECT username FROM users WHERE id = 1")
    user_result = cursor.fetchone()[0]
    print(f"Factory usage: Inserted and retrieved: {user_result}")
    conn.close()