testing-postgresql

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()