Skip to content

RexBytes/pgmonkey

BranchesTags

Repository files navigation

Getting Started with pgmonkey

pgmonkey is a Python library for managing PostgreSQL database connections. It supports normal, pooled, async, and async-pooled connections using a single YAML configuration file. Authentication methods include password, SSL/TLS, and certificate-based authentication. A CLI is included for managing configurations and testing connections.

Table of Contents

  1. Installation
  2. One Config File, All Connection Types
  3. YAML Configuration Reference
  4. Authentication Methods
  5. Environment Variable Interpolation (Advanced)
  6. Using the CLI
  7. Using pgmonkey in Python
  8. Best Practice Recipes
  9. Testing All Connection Types
  10. Testing Pool Capacity
  11. Running the Test Suite

Installation

Install from PyPI:

pip install pgmonkey

Or install from source:

git clone https://github.com/RexBytes/pgmonkey.git
cd pgmonkey
pip install .

To install with test dependencies:

pip install pgmonkey[test]

One Config File, All Connection Types

pgmonkey uses a single YAML configuration file for all connection types. Instead of maintaining separate config files for normal, pool, async, and async_pool connections, you define everything in one file and specify the connection type when you call the API:

from pgmonkey import PGConnectionManager

manager = PGConnectionManager()

# Same config file, different connection types
conn = manager.get_database_connection('config.yaml', 'normal')
conn = manager.get_database_connection('config.yaml', 'pool')
conn = await manager.get_database_connection('config.yaml', 'async')
conn = await manager.get_database_connection('config.yaml', 'async_pool')

The connection_type parameter is optional. If omitted, pgmonkey uses the connection_type value from the YAML file (which defaults to 'normal').

YAML Configuration Reference

Here is the full configuration template. You only need to fill in the sections relevant to the connection types you plan to use.

# Default connection type when none is specified in the API call.
# Options: 'normal', 'pool', 'async', 'async_pool'
# You can override this per-call:
#   manager.get_database_connection('config.yaml', 'pool')
connection_type: 'normal'

connection_settings:
  user: 'postgres'
  password: 'password'
  host: 'localhost'
  port: '5432'
  dbname: 'mydatabase'
  sslmode: 'prefer'  # Options: disable, allow, prefer, require, verify-ca, verify-full
  sslcert: ''  # Path to the client SSL certificate, if needed
  sslkey: ''  # Path to the client SSL key, if needed
  sslrootcert: ''  # Path to the root SSL certificate, if needed
  connect_timeout: '10'  # Maximum wait for connection, in seconds
  application_name: 'myapp'
  keepalives: '1'  # Enable TCP keepalives (1=on, 0=off)
  keepalives_idle: '60'  # Seconds before sending a keepalive probe
  keepalives_interval: '15'  # Seconds between keepalive probes
  keepalives_count: '5'  # Max keepalive probes before closing the connection

# Settings for 'pool' connection type
pool_settings:
  min_size: 5
  max_size: 20
  timeout: 30  # Seconds to wait for a connection from the pool before raising an error
  max_idle: 300  # Seconds a connection can remain idle before being closed
  max_lifetime: 3600  # Seconds a connection can be reused
  check_on_checkout: false  # Validate connections with SELECT 1 before handing to caller

# Settings for 'async' connection type (applied via SET commands on connection)
# These settings are also applied to 'async_pool' connections via a configure callback.
async_settings:
  idle_in_transaction_session_timeout: '5000'  # Timeout for idle in transaction (ms)
  statement_timeout: '30000'  # Cancel statements exceeding this time (ms)
  lock_timeout: '10000'  # Timeout for acquiring locks (ms)
  # work_mem: '256MB'  # Memory for sort operations and more

# Settings for 'async_pool' connection type
async_pool_settings:
  min_size: 5
  max_size: 20
  timeout: 30  # Seconds to wait for a connection from the pool before raising an error
  max_idle: 300
  max_lifetime: 3600
  check_on_checkout: false  # Validate connections with SELECT 1 before handing to caller

Connection Settings

Parameter Description Example
user Username for the PostgreSQL database 'postgres'
password Password for the database user 'password'
host Database server host address 'localhost'
port Database server port '5432'
dbname Name of the database to connect to 'mydatabase'
sslmode SSL mode (disable, allow, prefer, require, verify-ca, verify-full) 'prefer'
sslcert Path to the client SSL certificate '/path/to/client.crt'
sslkey Path to the client SSL key '/path/to/client.key'
sslrootcert Path to the root SSL certificate '/path/to/ca.crt'
connect_timeout Maximum wait for connection in seconds '10'
application_name Application name reported to PostgreSQL 'myapp'
keepalives Enable TCP keepalives (1=on, 0=off) '1'
keepalives_idle Seconds before sending a keepalive probe '60'
keepalives_interval Seconds between keepalive probes '15'
keepalives_count Max keepalive probes before closing '5'

Pool Settings

Used by pool connection type.

Parameter Description Example
min_size Minimum number of connections in the pool 5
max_size Maximum number of connections in the pool 20
timeout Seconds to wait for a connection from the pool before raising an error 30
max_idle Seconds a connection can remain idle before being closed 300
max_lifetime Seconds a connection can be reused 3600
check_on_checkout Validate connections with SELECT 1 before handing to caller false

Async Settings

Used by async and async_pool connection types. These are applied via SQL SET commands when the connection is established. For async_pool, they are applied to each connection via a psycopg_pool configure callback.

Parameter Description Example
idle_in_transaction_session_timeout Timeout for idle in transaction (ms) '5000'
statement_timeout Cancel statements exceeding this time (ms) '30000'
lock_timeout Timeout for acquiring locks (ms) '10000'
work_mem Memory for sort operations '256MB'

Async Pool Settings

Used by async_pool connection type. Same parameters as pool settings. The async_settings section (above) is also applied to async pool connections.

Parameter Description Example
min_size Minimum connections in the async pool 5
max_size Maximum connections in the async pool 20
timeout Seconds to wait for a connection from the pool before raising an error 30
max_idle Seconds a connection can remain idle 300
max_lifetime Seconds a connection can be reused 3600
check_on_checkout Validate connections with SELECT 1 before handing to caller false

Authentication Methods

Password-Based Authentication

The most common method. Credentials are sent to the PostgreSQL server for validation.

connection_type: 'normal'
connection_settings:
  user: 'your_user'
  password: 'your_password'
  host: 'localhost'
  dbname: 'your_database'

SSL/TLS Encryption

SSL/TLS encrypts the connection between your application and the PostgreSQL server. pgmonkey supports these SSL modes:

  • disable: No SSL.
  • allow: Attempt SSL, fall back to non-SSL if unavailable.
  • prefer: Attempt SSL, fall back to non-SSL if not supported.
  • require: Require SSL connection.
  • verify-ca: Require SSL and verify the server's certificate is signed by a trusted CA.
  • verify-full: Require SSL, verify certificate, and ensure the hostname matches.
connection_type: 'normal'
connection_settings:
  user: 'your_user'
  password: 'your_password'
  host: 'localhost'
  dbname: 'your_database'
  sslmode: 'require'
  sslrootcert: '/path/to/ca.crt'

Certificate-Based Authentication

Uses SSL client certificates for authentication. Highly secure and often used in enterprise environments.

connection_type: 'normal'
connection_settings:
  user: 'your_user'
  password: 'your_password'
  host: 'localhost'
  dbname: 'your_database'
  sslmode: 'verify-full'
  sslcert: '/path/to/client.crt'
  sslkey: '/path/to/client.key'
  sslrootcert: '/path/to/ca.crt'

Environment Variable Interpolation (Advanced)

pgmonkey v4.0.0 adds opt-in support for resolving environment variables and secret file references inside YAML configuration files. This lets you keep your config files free of hardcoded credentials while staying compatible with standard deployment workflows (12-factor env vars, Docker, Kubernetes).

Interpolation is disabled by default. If you do not enable it, pgmonkey treats every YAML value as a literal string - exactly as it always has.

Standard YAML (Default)

The standard approach is to write literal values directly in the config file. This is the simplest way to get started and requires no extra flags or API parameters:

connection_type: 'normal'

connection_settings:
  user: 'postgres'
  password: 'my_password'
  host: 'localhost'
  port: '5432'
  dbname: 'mydatabase'
from pgmonkey import PGConnectionManager

manager = PGConnectionManager()
conn = manager.get_database_connection('config.yaml')

This continues to work exactly as before. No changes needed for existing configs.

Inline Syntax: ${VAR}

When interpolation is enabled, you can reference environment variables using ${VAR} syntax. Use ${VAR:-default} to provide a fallback value when the variable is not set:

connection_type: 'normal'

connection_settings:
  user: '${PGUSER:-postgres}'
  password: '${PGPASSWORD}'
  host: '${PGHOST:-localhost}'
  port: '${PGPORT:-5432}'
  dbname: '${PGDATABASE:-mydb}'

Rules:

  • If a referenced variable is not set and no default is provided, pgmonkey raises EnvInterpolationError with a clear message naming the variable and the config key.
  • Multiple references can appear in a single value: '${PGHOST}:${PGPORT}'
  • Non-string values (integers, booleans) pass through unchanged.

Structured Syntax: from_env / from_file

For secrets, pgmonkey supports a structured YAML form that makes the intent unambiguous:

Read from an environment variable:

connection_settings:
  password:
    from_env: PGMONKEY_DB_PASSWORD

Read from a file (Kubernetes Secret-style):

connection_settings:
  password:
    from_file: /var/run/secrets/db/password

Rules:

  • from_env reads the named environment variable. If it is not set, pgmonkey raises EnvInterpolationError.
  • from_file reads the file contents and trims the trailing newline (matching Kubernetes Secret conventions). If the file does not exist or cannot be read, pgmonkey raises EnvInterpolationError.
  • A structured reference must be a dict with exactly one key (from_env or from_file). Dicts with additional keys are treated as normal nested config.

Sensitive Key Protection

By default, ${VAR:-default} fallback values are disallowed for sensitive keys. This prevents accidentally shipping a config with a hardcoded fallback password that silently takes over when the env var is missing.

Sensitive keys: password, sslkey, sslcert, sslrootcert, and any key containing token, secret, or credential.

# This will FAIL (password is a sensitive key, defaults not allowed):
password: '${PGPASSWORD:-fallback_password}'

# This is fine (host is not sensitive):
host: '${PGHOST:-localhost}'

To explicitly allow sensitive defaults (e.g. for local development), pass allow_sensitive_defaults=True:

cfg = load_config('config.yaml', resolve_env=True, allow_sensitive_defaults=True)

Python API

Use load_config() for the simplest path to a resolved config dictionary:

from pgmonkey import load_config

# Load without interpolation (default - same as always)
cfg = load_config('config.yaml')

# Load with env interpolation enabled
cfg = load_config('config.yaml', resolve_env=True)

# Strict mode: fail on missing vars
cfg = load_config('config.yaml', resolve_env=True, strict=True)

Or pass resolve_env=True directly to the connection manager:

from pgmonkey import PGConnectionManager

manager = PGConnectionManager()
conn = manager.get_database_connection('config.yaml', resolve_env=True)

Both get_database_connection() and get_database_connection_from_dict() accept the resolve_env parameter.

CLI Usage

Add --resolve-env to any pgconfig test or pgconfig generate-code command:

# Test a connection with env vars resolved
pgmonkey pgconfig test --connconfig config.yaml --resolve-env

# Generate code (--resolve-env accepted for consistency)
pgmonkey pgconfig generate-code --connconfig config.yaml --resolve-env

Without --resolve-env, the CLI works exactly as before - ${VAR} patterns are treated as literal strings.

Redacting Secrets for Logging

pgmonkey includes a redact_config() utility that replaces sensitive values with ***REDACTED***, safe for logging or printing:

from pgmonkey import load_config
from pgmonkey.common.utils.redaction import redact_config

cfg = load_config('config.yaml', resolve_env=True)
print(redact_config(cfg))
# {'connection_settings': {'password': '***REDACTED***', 'host': 'db.prod.com', ...}}

Redacted keys: password, sslkey, sslcert, sslrootcert, and any key containing token, secret, or credential. Empty strings and None values are left as-is (nothing to leak).

Deployment Patterns

Local development - set env vars in your shell or a .env file (managed by your own tooling):

export PGPASSWORD=dev_password
export PGHOST=localhost

Docker / containers - pass env vars via docker run -e or docker-compose.yml:

# docker-compose.yml
services:
  app:
    environment:
      PGPASSWORD: ${DB_PASSWORD}
      PGHOST: db

Kubernetes - mount secrets as files and use from_file:

# pgmonkey config
connection_settings:
  password:
    from_file: /var/run/secrets/db/password
  host: '${PGHOST:-db-service}'
# k8s pod spec
volumes:
  - name: db-secret
    secret:
      secretName: db-credentials
containers:
  - volumeMounts:
      - name: db-secret
        mountPath: /var/run/secrets/db
        readOnly: true

Cloud secret managers (AWS Secrets Manager, GCP Secret Manager, HashiCorp Vault) - resolve secrets in your deployment pipeline and set them as env vars. pgmonkey does not integrate with cloud vaults directly.

Using the CLI

pgmonkey provides a command-line interface for managing configurations and connections.

pgmonkey --help

Creating a Configuration Template

Generate a YAML configuration template:

pgmonkey pgconfig create --type pg --filepath /path/to/config.yaml

This creates a configuration file with all available settings and sensible defaults. Edit the file to customize your connection settings.

Testing a Connection

Test a connection using your configuration file:

# Test using the connection_type from the config file
pgmonkey pgconfig test --connconfig /path/to/config.yaml

# Test a specific connection type (overrides config file)
pgmonkey pgconfig test --connconfig /path/to/config.yaml --connection-type pool
pgmonkey pgconfig test --connconfig /path/to/config.yaml --connection-type async

The --connection-type flag accepts: normal, pool, async, async_pool.

Generating Python Code

Generate example Python code for a connection type:

# Generate code using the config file's default connection type
pgmonkey pgconfig generate-code --filepath /path/to/config.yaml

# Generate code for a specific connection type
pgmonkey pgconfig generate-code --filepath /path/to/config.yaml --connection-type async_pool

# Generate code using native psycopg/psycopg_pool instead of pgmonkey
pgmonkey pgconfig generate-code --filepath /path/to/config.yaml --connection-type pool --library psycopg

The --library flag controls which library the generated code targets:

  • pgmonkey (default) - generates code using pgmonkey's PGConnectionManager.
  • psycopg - generates code using psycopg and psycopg_pool directly, reading connection settings from the same YAML config file.

Server Configuration Recommendations

Generate recommended PostgreSQL server configuration entries based on your config file:

pgmonkey pgserverconfig --filepath /path/to/config.yaml

This analyzes your configuration and outputs recommended entries for postgresql.conf and pg_hba.conf:

1) Database type detected: PostgreSQL

2) Minimal database server settings needed for this config file:

   a) pg_hba.conf:

TYPE  DATABASE  USER  ADDRESS          METHOD  OPTIONS
hostssl all     all   192.168.0.0/24   md5     clientcert=verify-full

   b) postgresql.conf:

max_connections = 22
ssl = on
ssl_cert_file = 'server.crt'
ssl_key_file = 'server.key'
ssl_ca_file = 'ca.crt'

Auditing Live Server Settings

Add --audit to connect to the live server and compare current settings against recommendations:

pgmonkey pgserverconfig --filepath /path/to/config.yaml --audit

This queries the server's pg_settings (read-only) and displays a comparison table:

1) Database type detected: PostgreSQL

2) Server settings audit:

   postgresql.conf:

   Setting          Recommended  Current  Source              Status
   ────────────────────────────────────────────────────────────────────
   max_connections   22           100      configuration file  OK
   ssl               on           on       configuration file  OK
   ssl_cert_file     server.crt   server.crt  configuration file  OK
   ssl_key_file      server.key   server.key  configuration file  OK
   ssl_ca_file       ca.crt       ca.crt      configuration file  OK

The audit also inspects pg_hba_file_rules (PostgreSQL 15+) when available, showing current HBA rules alongside recommendations.

If the connected role lacks permission to query pg_settings, the audit fails gracefully with a message and falls back to showing recommendations only. No server settings are ever modified - the audit is entirely read-only.

Importing and Exporting Data

Import data from a CSV or text file into a PostgreSQL table:

pgmonkey pgimport --table public.my_table --connconfig /path/to/config.yaml --import_file /path/to/data.csv

If an import configuration file doesn't exist, pgmonkey generates a template you can edit to adjust column mapping, delimiter, and encoding.

Export data from a PostgreSQL table to a CSV file:

pgmonkey pgexport --table public.my_table --connconfig /path/to/config.yaml --export_file /path/to/output.csv

If --export_file is omitted, a default file is generated using the table name.

Using pgmonkey in Python

Normal (Synchronous) Connection

from pgmonkey import PGConnectionManager

def main():
    manager = PGConnectionManager()
    connection = manager.get_database_connection('config.yaml', 'normal')

    with connection as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT version();')
            print(cur.fetchone())

if __name__ == "__main__":
    main()

Pooled Connection

from pgmonkey import PGConnectionManager

def main():
    manager = PGConnectionManager()
    pool_connection = manager.get_database_connection('config.yaml', 'pool')

    # Each 'with' block acquires and releases a connection from the pool
    with pool_connection as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT version();')
            print(cur.fetchone())

if __name__ == "__main__":
    main()

Async Connection

import asyncio
from pgmonkey import PGConnectionManager

async def main():
    manager = PGConnectionManager()
    connection = await manager.get_database_connection('config.yaml', 'async')

    async with connection as conn:
        async with conn.cursor() as cur:
            await cur.execute('SELECT version();')
            print(await cur.fetchone())

if __name__ == "__main__":
    asyncio.run(main())

Async Pooled Connection

import asyncio
from pgmonkey import PGConnectionManager

async def main():
    manager = PGConnectionManager()
    pool_connection = await manager.get_database_connection('config.yaml', 'async_pool')

    # Each 'async with' cursor block acquires and releases a connection from the pool
    async with pool_connection as conn:
        async with conn.cursor() as cur:
            await cur.execute('SELECT version();')
            print(await cur.fetchone())

if __name__ == "__main__":
    asyncio.run(main())

Using the Config File Default

If you omit the connection_type parameter, pgmonkey uses the value from your YAML file:

# Uses whatever connection_type is set in config.yaml (defaults to 'normal')
connection = manager.get_database_connection('config.yaml')

Transactions, Commit, and Rollback

pgmonkey connections support transactions via context managers:

# Synchronous transaction
with connection as conn:
    with conn.transaction():
        with conn.cursor() as cur:
            cur.execute('INSERT INTO my_table (name) VALUES (%s)', ('Alice',))
            cur.execute('SELECT * FROM my_table WHERE name = %s', ('Alice',))
            print(cur.fetchall())

# Asynchronous transaction
async with connection as conn:
    async with conn.transaction():
        async with conn.cursor() as cur:
            await cur.execute('INSERT INTO my_table (name) VALUES (%s)', ('Alice',))
            await cur.execute('SELECT * FROM my_table WHERE name = %s', ('Alice',))
            print(await cur.fetchall())

Manual commit and rollback are available when not using the transaction context:

# Manual commit
async with connection as conn:
    async with conn.cursor() as cur:
        await cur.execute('UPDATE my_table SET name = %s WHERE id = %s', ('Doe', 1))
    await conn.commit()

# Manual rollback on error
try:
    async with connection as conn:
        async with conn.cursor() as cur:
            await cur.execute('DELETE FROM my_table WHERE id = %s', (1,))
        await conn.commit()
except Exception as e:
    await conn.rollback()

Best Practice Recipes

pgmonkey handles several production concerns behind the scenes so you don't have to:

  • Connection caching - Connections and pools are cached by config content (SHA-256 hash). Repeated calls with the same config return the existing instance, preventing "pool storms" where each call opens a new pool.
  • Async pool lifecycle - async with pool_conn: borrows a connection from the pool and returns it when the block exits. The pool stays open for reuse. Auto-commits on clean exit, rolls back on exception.
  • atexit cleanup - All cached connections are automatically closed when the process exits.
  • Thread-safe caching - The connection cache is protected by a threading lock with double-check locking to prevent race conditions.
  • Config validation - Unknown connection setting keys produce a warning log message. Pool settings are validated (e.g., min_size cannot exceed max_size).

App-Level Pattern: Sync Database Class (Flask)

from pgmonkey import PGConnectionManager

class Database:
    def __init__(self, config_path):
        self.manager = PGConnectionManager()
        self.config_path = config_path
        # Pool is created on first call, cached thereafter
        self.pool = self.manager.get_database_connection(config_path, 'pool')

    def fetch_one(self, query, params=None):
        with self.pool as conn:
            with conn.cursor() as cur:
                cur.execute(query, params)
                return cur.fetchone()

    def fetch_all(self, query, params=None):
        with self.pool as conn:
            with conn.cursor() as cur:
                cur.execute(query, params)
                return cur.fetchall()

    def execute(self, query, params=None):
        with self.pool as conn:
            with conn.cursor() as cur:
                cur.execute(query, params)

# Usage in Flask
from flask import Flask

app = Flask(__name__)
db = Database('/path/to/config.yaml')

@app.route('/users')
def list_users():
    rows = db.fetch_all('SELECT id, name FROM users ORDER BY id;')
    return {'users': [{'id': r[0], 'name': r[1]} for r in rows]}

App-Level Pattern: Async Database Class (FastAPI)

import asyncio
from pgmonkey import PGConnectionManager

class AsyncDatabase:
    def __init__(self, config_path):
        self.manager = PGConnectionManager()
        self.config_path = config_path
        self.pool = None

    async def connect(self):
        self.pool = await self.manager.get_database_connection(
            self.config_path, 'async_pool'
        )

    async def disconnect(self):
        await self.manager.clear_cache_async()

    async def fetch_one(self, query, params=None):
        async with self.pool as conn:
            async with conn.cursor() as cur:
                await cur.execute(query, params)
                return await cur.fetchone()

    async def fetch_all(self, query, params=None):
        async with self.pool as conn:
            async with conn.cursor() as cur:
                await cur.execute(query, params)
                return await cur.fetchall()

    async def execute(self, query, params=None):
        async with self.pool as conn:
            async with conn.cursor() as cur:
                await cur.execute(query, params)

# Usage in FastAPI
from fastapi import FastAPI

app = FastAPI()
db = AsyncDatabase('/path/to/config.yaml')

@app.on_event("startup")
async def startup():
    await db.connect()

@app.on_event("shutdown")
async def shutdown():
    await db.disconnect()

@app.get("/orders")
async def list_orders():
    rows = await db.fetch_all('SELECT id, total FROM orders ORDER BY id;')
    return {"orders": [{"id": r[0], "total": r[1]} for r in rows]}

Cache Management

Method Description
manager.cache_info Returns dict with size and connection_types of cached connections
manager.clear_cache() Disconnects all cached connections (sync)
await manager.clear_cache_async() Disconnects all cached connections (async)
force_reload=True Pass to get_database_connection() to replace a cached connection

Quick Reference

Type Best For Cached? Context Manager
normal Scripts, CLI tools Yes with conn:
pool Flask, Django, threaded apps Yes with pool: borrows/returns
async Async scripts Yes async with conn:
async_pool FastAPI, aiohttp, high concurrency Yes async with pool: borrows/returns

For full recipes with code examples for every connection type, see the Best Practices documentation page.

Testing All Connection Types

Test all four connection types using a single config file:

import asyncio
from pgmonkey import PGConnectionManager

def test_sync(manager, config_file, connection_type):
    connection = manager.get_database_connection(config_file, connection_type)
    with connection as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT version();')
            print(f"{connection_type}: {cur.fetchone()}")

async def test_async(manager, config_file, connection_type):
    connection = await manager.get_database_connection(config_file, connection_type)
    async with connection as conn:
        async with conn.cursor() as cur:
            await cur.execute('SELECT version();')
            print(f"{connection_type}: {await cur.fetchone()}")

async def main():
    manager = PGConnectionManager()
    config_file = '/path/to/config.yaml'

    # Test synchronous connections
    test_sync(manager, config_file, 'normal')
    test_sync(manager, config_file, 'pool')

    # Test asynchronous connections
    await test_async(manager, config_file, 'async')
    await test_async(manager, config_file, 'async_pool')

if __name__ == "__main__":
    asyncio.run(main())

Testing Pool Capacity

Test pooling by acquiring multiple connections from the same pool:

import asyncio
from pgmonkey import PGConnectionManager

async def test_async_pool(config_file, num_connections):
    manager = PGConnectionManager()
    connections = []

    for _ in range(num_connections):
        connection = await manager.get_database_connection(config_file, 'async_pool')
        connections.append(connection)

    for idx, connection in enumerate(connections):
        async with connection as conn:
            async with conn.cursor() as cur:
                await cur.execute('SELECT version();')
                version = await cur.fetchone()
                print(f"Async pool connection {idx + 1}: {version}")

    for connection in connections:
        await connection.disconnect()

def test_sync_pool(config_file, num_connections):
    manager = PGConnectionManager()
    connections = []

    for _ in range(num_connections):
        connection = manager.get_database_connection(config_file, 'pool')
        connections.append(connection)

    for idx, connection in enumerate(connections):
        with connection as conn:
            with conn.cursor() as cur:
                cur.execute('SELECT version();')
                version = cur.fetchone()
                print(f"Sync pool connection {idx + 1}: {version}")

    for connection in connections:
        connection.disconnect()

async def main():
    config_file = '/path/to/config.yaml'
    num_connections = 5

    print("Testing async pool:")
    await test_async_pool(config_file, num_connections)

    print("\nTesting sync pool:")
    test_sync_pool(config_file, num_connections)

if __name__ == "__main__":
    asyncio.run(main())

Running the Test Suite

pgmonkey includes a comprehensive unit test suite that runs without a database connection.

Install test dependencies:

pip install pgmonkey[test]

Run the tests:

pytest

The test suite uses mocks and covers all connection types, the connection factory, configuration management, code generation (both pgmonkey and native psycopg), config validation, server config generation, and server settings inspection.

For more information, visit the GitHub repository.

About

No description, website, or topics provided.

Resources

Readme

License

MIT license
Activity

Stars

1 star

Watchers

1 watching

Forks

0 forks
Report repository

Releases 30

# pgmonkey v4.0.0 Release Notes Latest
Feb 17, 2026
+ 29 releases

Packages

 
 
 

Contributors

Languages