PyPGKit

A PostgreSQL database framework for Python using psycopg3 with connection pooling, thread-safety, and repository pattern.

Features

• Connection Pooling: Thread-safe connection pool using psycopg_pool
• Singleton Pattern: Global database instance with Database.init() and Database.get_instance()
• Repository Pattern: Base repository class with CRUD operations
• Automatic Setup: Create database, user, and schema automatically
• Fork Safety: Compatible with gunicorn and other forking servers
• Configurable Logging: Built-in logging with multiple formats
• Type Hints: Full type annotations for better IDE support

Installation

pip install PyPGKit

Quick Start

Basic Usage

from pypgkit import Database, DatabaseConfig

# Initialize once at application startup
config = DatabaseConfig(
    host="localhost",
    port=5432,
    database="myapp",
    user="myuser",
    password="mypassword",
)

# This creates the database/user if needed and initializes the connection pool
Database.init(config=config, schema_path="schemas/init.sql")

# Use anywhere in your application
db = Database.get_instance()

# Execute queries
users = db.fetch_all("SELECT * FROM users WHERE active = %s", (True,))

Environment Variables

pypgkit supports configuration via environment variables:

export PYPGKIT_HOST=localhost
export PYPGKIT_PORT=5432
export PYPGKIT_DATABASE=myapp
export PYPGKIT_USER=myuser
export PYPGKIT_PASSWORD=mypassword

from pypgkit import Database, DatabaseConfig

# Load from environment
config = DatabaseConfig.from_env()
Database.init(config=config)

Repository Pattern

Create type-safe repositories for your entities:

from dataclasses import dataclass
from typing import Optional
from pypgkit import Database
from pypgkit.repositories import BaseRepository

@dataclass
class User:
    id: Optional[int] = None
    email: str = ""
    name: str = ""
    active: bool = True

class UserRepository(BaseRepository[User]):
    table_name = "users"
    primary_key = "id"

    def _row_to_entity(self, row: dict) -> User:
        return User(
            id=row.get("id"),
            email=row.get("email", ""),
            name=row.get("name", ""),
            active=row.get("active", True),
        )

    def _entity_to_row(self, entity: User) -> dict:
        row = {"email": entity.email, "name": entity.name, "active": entity.active}
        if entity.id is not None:
            row["id"] = entity.id
        return row

# Usage
db = Database.get_instance()
repo = UserRepository(db)

# Create
user = repo.create(User(email="user@example.com", name="John"))

# Read
user = repo.find_by_id(1)
users = repo.find_all(limit=10)
active_users = repo.find_by({"active": True})

# Update
user.name = "Jane"
repo.update(user)

# Delete
repo.delete(1)

Logging Configuration

from pypgkit.logging import configure_logging, LogLevel, DETAILED_FORMAT

# Basic configuration
configure_logging(level=LogLevel.INFO)

# Debug mode with detailed format
configure_logging(level=LogLevel.DEBUG, format=DETAILED_FORMAT)

# Log to file
configure_logging(level=LogLevel.INFO, filename="pypgkit.log")

# Include psycopg logs
configure_logging(level=LogLevel.DEBUG, include_psycopg=True)

Automatic Database Setup

pypgkit can automatically create the database and user if they don't exist:

from pypgkit import Database, DatabaseConfig

config = DatabaseConfig(
    database="myapp",
    user="myuser",
    password="mypassword",
)

# Interactive mode (prompts for admin credentials)
Database.init(config=config, auto_setup=True, interactive=True)

# Non-interactive mode (for CI/CD)
from pypgkit.setup import setup_database
setup_database(
    config,
    admin_user="postgres",
    admin_password="adminpass",
    interactive=False,
)

Using with Gunicorn

pypgkit automatically handles fork detection for compatibility with gunicorn:

# wsgi.py
from pypgkit import Database, DatabaseConfig

config = DatabaseConfig.from_env()
Database.init(config=config)

# The connection pool automatically resets after fork

gunicorn -w 4 myapp:app

API Reference

DatabaseConfig

Configuration dataclass for database connections.

Database

Main database facade class.

BaseRepository

Base class for repositories with CRUD operations.

Development

Setup

git clone https://github.com/miichoow/PyPGKit.git
cd PyPGKit
pip install -e ".[dev]"

Running Tests

pytest tests/ -v --cov=pypgkit --cov-report=term-missing

Linting

ruff check pypgkit/
ruff format pypgkit/

License

MIT License - see LICENSE file for details.