aa_psycopg 0.1.4

Internal Partenamut library for PostgreSQL access using Psycopg 3

aa-psycopg

Internal Partenamut library for PostgreSQL access using Psycopg 3.
Provides a simple and efficient interface for interacting with PostgreSQL using:

• A direct connection (PostgreSQLConnection)
• A connection pool (PostgreSQLPool)

Supports query execution (SELECT, INSERT, UPDATE, DELETE), transactions, schema caching, and query runtime statistics.

---

Installation

pip install aa-psycopg

Requirements

• Python 3.11+
• psycopg 3
• psycopg_pool

---

Features

• Easy-to-use wrapper for psycopg3.
• Connection pooling via psycopg_pool.
• Automatic query runtime tracking.
• Safe connection string handling (masks passwords in logs).
• Schema caching for executed queries.

---

API overview

• ping(retries=0, timeout=60, query_name="ping") -> bool
  Test database connectivity.
• read(query, params=None, query_name=None) -> list[dict]
  Execute a SELECT query. Set query_name to track query runtime statistics.
• write(query, params=None, returning=False, query_name=None) -> list[dict] | None
  Execute INSERT/UPDATE/DELETE (optionally returning results). Set query_name to track query runtime statistics.
• execute_transaction(queries_params, query_name=None)
  Run multiple queries inside a transaction. Set query_name to track query runtime statistics.
• get_stats() -> dict
  Retrieve runtime stats (execution time & call count per query_name).
• get_schema(query) -> list[psycopg.Column]
  Get schema for a previously executed query.

Usage

Direct Connection

import logging
import json

from aa_psycopg.connection import PostgreSQLConnection

# Configure logging (can adjust level or handlers as needed)
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

with PostgreSQLConnection(
    user="myuser",
    password="mypassword",
    host="localhost",
    port=5432,
    db="mydatabase"
) as client:
    # Check if database is alive
    client.ping()

    # Read data
    rows = client.read("SELECT * FROM my_table", query_name="fetch_all")
    logger.info("Fetched rows: %s", rows)

    # Write data
    client.write(
        "INSERT INTO my_table (name) VALUES (%(name)s)",
        params={"name": "example"},
        query_name="insert_row"
    )
    logger.info("Inserted row into my_table")

    # Transaction
    client.execute_transaction([
        ("INSERT INTO my_table (name) VALUES (%(name)s)", {"name": "row1"}),
        ("INSERT INTO my_table (name) VALUES (%(name)s)", {"name": "row2"}),
    ], query_name="bulk_insert")
    logger.info("Executed bulk insert transaction")

    # Query stats
    stats = client.get_stats()
    logger.info("Query stats: %s", json.dumps(stats, indent=4))

---

Connection Pool

import logging
import json
from aa_psycopg.pool import PostgreSQLPool

# Configure logging (can be redirected to CloudWatch or file)
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

with PostgreSQLPool(
    user="myuser",
    password="mypassword",
    host="localhost",
    port=5432,
    db="mydatabase",
    min_size=1,
    max_size=5
) as client:
    # Check connectivity
    client.ping()

    # Fetch results
    results = client.read("SELECT * FROM users WHERE active = true", query_name="active_users")
    logger.info("Fetched results: %s", results)

    # Insert with RETURNING
    new_ids = client.write(
        "INSERT INTO users (name) VALUES (%(name)s) RETURNING id",
        params={"name": "Alice"},
        returning=True,
        query_name="insert_user"
    )
    logger.info("Inserted new user IDs: %s", new_ids)

    # Bulk transaction
    client.execute_transaction([
        ("UPDATE users SET active=false WHERE id=%(id)s", {"id": 1}),
        ("DELETE FROM users WHERE active=false", None),
    ], query_name="cleanup")
    logger.info("Executed cleanup transaction")

    # Stats (dump as JSON)
    stats = client.get_stats()
    logger.info("Query stats: %s", json.dumps(stats, indent=4))

---

Contributing

Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

Please make sure to update tests as appropriate.

---