aa_psycopg 0.1.3

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

from aa_psycopg.connection import PostgreSQLConnection
with PostgreSQLConnection(
    user="myuser",
    password="mypassword",
    host="localhost",
    port=5432,
    db="mydatabase"
) as client:
    # Check if database is alive
    if client.ping():
        print("Database reachable!")
    # Read data
    rows = client.read("SELECT * FROM my_table", query_name="fetch_all")
    print(rows)
    # Write data
    client.write(
        "INSERT INTO my_table (name) VALUES (%(name)s)",
        params={"name": "example"},
        query_name="insert_row"
    )
    # 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")
    # Query stats
    print(client.get_stats())

---

Connection Pool

from aa_psycopg.pool import PostgreSQLPool
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")
    print(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"
    )
    print(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")
    # Stats
    print(client.get_stats())

---

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.

---