pg_helper

A simple and robust helper package for PostgreSQL operations. This package provides an easy-to-use interface for connecting to PostgreSQL databases, executing queries, and handling results with various fetching modes.

Features

• Connection Management: Supports both managed (context manager) and unmanaged connection modes with configurable auto-commit.
• Flexible Querying: Execute SQL queries with support for parameterized queries.
• Row Styles: Fetch results as dictionaries or tuples.
• Advanced Fetching: Multiple fetch modes including column extraction, key-value pairs, grouping, and more.
• Bulk Operations: Efficiently execute the same query multiple times with different parameters using execute_many.
• Transaction Control: Manual commit and rollback support.

Installation

Install the package using pip:

pip install pg_helper

Requirements

• Python >= 3.8
• psycopg[binary] >= 3.1

Usage

Basic Connection and Querying

from pg_helper import DBHelper, RowStyle

# Using environment variable POSTGRESQL_URL for connection string
with DBHelper() as db:
    result = db.query("SELECT * FROM users WHERE active = %s", (True,))
    for user in result:
        print(user['name'])  # Assuming RowStyle.DICT (default)

# Or pass connection string directly
db = DBHelper("postgresql://user:password@localhost/dbname")
try:
    result = db.query("SELECT id, name FROM products", style=RowStyle.TUPLE)
    products = result.fetch_all()
    print(products)
finally:
    db.close()

Fetching Modes

from pg_helper import DBHelper, FetchMode

with DBHelper() as db:
    result = db.query("SELECT category, name, price FROM products")

    # Get all categories as a list
    categories = result.fetch_all(FetchMode.COLUMN, key_column='category')

    # Get products grouped by category
    products_by_category = result.fetch_all(FetchMode.GROUP, key_column='category')

    # Get price mapping by product name
    price_map = result.fetch_all(FetchMode.KEY_PAIR, key_column='name', value_column='price')

Bulk Insert

from pg_helper import DBHelper

users_data = [
    ('Alice', 'alice@example.com'),
    ('Bob', 'bob@example.com'),
    ('Charlie', 'charlie@example.com')
]

with DBHelper() as db:
    affected_rows = db.execute_many(
        "INSERT INTO users (name, email) VALUES (%s, %s)",
        users_data
    )
    print(f"Inserted {affected_rows} users")

API Reference

DBHelper

The main class for database operations.

• __init__(conn_str=None, autocommit=False): Initialize with optional connection string and auto-commit mode.
• query(query, params=None, style=RowStyle.DICT): Execute a query and return a QueryResult object.
• execute_many(query, param_list): Execute a query multiple times with different parameters.
• commit(): Manually commit the current transaction.
• rollback(): Manually roll back the current transaction.
• close(): Close the database connection.

QueryResult

An iterable container for query results.

• fetch(style=None): Fetch the next row.
• fetch_all(mode=None, key_column=0, value_column=1): Fetch all remaining rows with optional mode.
• row_count: Property returning the number of affected rows.

Enums

• RowStyle: DICT, TUPLE
• FetchMode: COLUMN, KEY_PAIR, KEY_PAIR_LIST, GROUP, GROUP_COLUMN