# Psycopg Toolkit
[](https://github.com/descoped/psycopg-toolkit/actions/workflows/build-test-native.yml)
[](https://codecov.io/gh/descoped/psycopg-toolkit)
[](https://www.python.org/downloads/)
[](https://opensource.org/licenses/MIT)
[](https://github.com/descoped/psycopg-toolkit/releases)
A robust PostgreSQL database toolkit providing enterprise-grade connection pooling and database management capabilities for Python applications.
## Features
- Async-first design with connection pooling via `psycopg-pool`
- Comprehensive transaction management with savepoint support
- Type-safe repository pattern with Pydantic model validation
- JSONB support with automatic field detection and psycopg JSON adapters
- SQL query builder with SQL injection protection
- Database schema and test data lifecycle management
- Automatic retry mechanism with exponential backoff
- Granular exception hierarchy for error handling
- Connection health monitoring and validation
- Database initialization callback system
- Statement timeout configuration
- Fully typed with modern Python type hints
## Installation
```bash
pip install psycopg-toolkit
```
## Quick Start
```python
from psycopg_toolkit import Database, DatabaseSettings
from uuid import uuid4
# Configure database
settings = DatabaseSettings(
host="localhost",
port=5432,
dbname="your_database",
user="your_user",
password="your_password"
)
async def main():
# Initialize database
db = Database(settings)
await db.init_db()
# Get transaction manager
tm = await db.get_transaction_manager()
# Execute in transaction
async with tm.transaction() as conn:
async with conn.cursor() as cur:
user_id = uuid4()
await cur.execute(
"INSERT INTO users (id, email) VALUES (%s, %s)",
(user_id, "user@example.com")
)
# Clean up
await db.cleanup()
```
## Core Components
### Database Management
```python
# Health check
is_healthy = await db.check_pool_health()
# Connection management
async with db.connection() as conn:
async with conn.cursor() as cur:
await cur.execute("SELECT version()")
```
### Transaction Management
```python
# Basic transaction
async with tm.transaction() as conn:
# Operations automatically rolled back on error
pass
# With savepoint
async with tm.transaction(savepoint="user_creation") as conn:
# Nested transaction using savepoint
pass
```
### Repository Pattern
```python
from pydantic import BaseModel
from psycopg_toolkit import BaseRepository
class User(BaseModel):
id: UUID
email: str
class UserRepository(BaseRepository[User]):
def __init__(self, conn: AsyncConnection):
super().__init__(
db_connection=conn,
table_name="users",
model_class=User,
primary_key="id"
)
# Usage
async with tm.transaction() as conn:
repo = UserRepository(conn)
user = await repo.get_by_id(user_id)
```
### JSONB Support
```python
from typing import Dict, List, Any
from pydantic import BaseModel
from psycopg_toolkit import BaseRepository
class UserProfile(BaseModel):
id: int
name: str
# These fields are automatically detected as JSONB
metadata: Dict[str, Any]
preferences: Dict[str, str]
tags: List[str]
class UserRepository(BaseRepository[UserProfile, int]):
def __init__(self, conn):
super().__init__(
db_connection=conn,
table_name="user_profiles",
model_class=UserProfile,
primary_key="id"
# auto_detect_json=True by default
)
# Usage - JSON fields handled automatically
user = UserProfile(
id=1,
name="John Doe",
metadata={"created_at": "2024-01-01", "source": "web"},
preferences={"theme": "dark", "language": "en"},
tags=["premium", "beta_tester"]
)
# JSONB fields automatically serialized/deserialized
created_user = await repo.create(user)
retrieved_user = await repo.get_by_id(1)
```
### Schema Management
```python
from psycopg_toolkit.core.transaction import SchemaManager
class UserSchemaManager(SchemaManager[None]):
async def create_schema(self, conn: AsyncConnection) -> None:
await conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id UUID PRIMARY KEY,
email TEXT UNIQUE NOT NULL
)
""")
async def drop_schema(self, conn: AsyncConnection) -> None:
await conn.execute("DROP TABLE IF EXISTS users")
# Usage
async with tm.with_schema(UserSchemaManager()) as _:
# Schema available here
pass # Automatically dropped after
```
## Error Handling
```python
from psycopg_toolkit import (
DatabaseConnectionError,
DatabasePoolError,
DatabaseNotAvailable,
RecordNotFoundError
)
try:
async with tm.transaction() as conn:
repo = UserRepository(conn)
user = await repo.get_by_id(user_id)
except DatabaseConnectionError as e:
print(f"Connection error: {e.original_error}")
except RecordNotFoundError:
print(f"User {user_id} not found")
```
## Documentation
- [Database Management](docs/database.md)
- [Transaction Management](docs/transaction_manager.md)
- [Base Repository](docs/base_repository.md)
- [JSONB Support](docs/jsonb_support.md)
- [PsycopgHelper](docs/psycopg_helper.md)
## Running Tests
```bash
# Install dependencies
uv sync --all-groups
# Run all tests except performance tests (default)
uv run pytest
# Run only performance tests
uv run pytest -m performance
# Run all tests including performance
uv run pytest -m ""
# Run specific test categories
uv run pytest tests/unit/ # Only unit tests
uv run pytest -m performance # Only performance tests
# Run with coverage
uv run pytest --cov=src/psycopg_toolkit --cov-report=html
```
### Test Categories
The test suite is organized into three categories:
- **Unit tests**: Fast, isolated tests that don't require a database (in `tests/unit/`)
- **Integration tests**: Tests that require a real PostgreSQL database (in `tests/` root)
- **Performance tests**: Benchmarks and performance measurements (marked with `@pytest.mark.performance`)
Performance tests are excluded by default to keep the regular test runs fast. Use the `-m performance` flag to run them explicitly.
## Contributing
1. Fork the repository
2. Create a feature branch
3. Add tests for new features
4. Ensure all tests pass
5. Submit a pull request
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "psycopg-toolkit",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "async, database, jsonb, pool, postgresql, psycopg, repository-pattern",
"author": null,
"author_email": "Ove Ranheim <oranheim@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/78/5d/0b921f3b1624ee5f3eb37516c12406f4a7f0f2f4d5195908b1bffe8acd7f/psycopg_toolkit-0.2.0.tar.gz",
"platform": null,
"description": "# Psycopg Toolkit\n\n[](https://github.com/descoped/psycopg-toolkit/actions/workflows/build-test-native.yml)\n[](https://codecov.io/gh/descoped/psycopg-toolkit)\n[](https://www.python.org/downloads/)\n[](https://opensource.org/licenses/MIT)\n[](https://github.com/descoped/psycopg-toolkit/releases)\n\nA robust PostgreSQL database toolkit providing enterprise-grade connection pooling and database management capabilities for Python applications.\n\n## Features\n\n- Async-first design with connection pooling via `psycopg-pool`\n- Comprehensive transaction management with savepoint support\n- Type-safe repository pattern with Pydantic model validation\n- JSONB support with automatic field detection and psycopg JSON adapters\n- SQL query builder with SQL injection protection\n- Database schema and test data lifecycle management\n- Automatic retry mechanism with exponential backoff\n- Granular exception hierarchy for error handling\n- Connection health monitoring and validation\n- Database initialization callback system\n- Statement timeout configuration\n- Fully typed with modern Python type hints\n\n## Installation\n\n```bash\npip install psycopg-toolkit\n```\n\n## Quick Start\n\n```python\nfrom psycopg_toolkit import Database, DatabaseSettings\nfrom uuid import uuid4\n\n# Configure database\nsettings = DatabaseSettings(\n host=\"localhost\",\n port=5432,\n dbname=\"your_database\",\n user=\"your_user\",\n password=\"your_password\"\n)\n\nasync def main():\n # Initialize database\n db = Database(settings)\n await db.init_db()\n \n # Get transaction manager\n tm = await db.get_transaction_manager()\n \n # Execute in transaction\n async with tm.transaction() as conn:\n async with conn.cursor() as cur:\n user_id = uuid4()\n await cur.execute(\n \"INSERT INTO users (id, email) VALUES (%s, %s)\",\n (user_id, \"user@example.com\")\n )\n \n # Clean up\n await db.cleanup()\n```\n\n## Core Components\n\n### Database Management\n\n```python\n# Health check\nis_healthy = await db.check_pool_health()\n\n# Connection management\nasync with db.connection() as conn:\n async with conn.cursor() as cur:\n await cur.execute(\"SELECT version()\")\n```\n\n### Transaction Management\n\n```python\n# Basic transaction\nasync with tm.transaction() as conn:\n # Operations automatically rolled back on error\n pass\n\n# With savepoint\nasync with tm.transaction(savepoint=\"user_creation\") as conn:\n # Nested transaction using savepoint\n pass\n```\n\n### Repository Pattern\n\n```python\nfrom pydantic import BaseModel\nfrom psycopg_toolkit import BaseRepository\n\nclass User(BaseModel):\n id: UUID\n email: str\n\nclass UserRepository(BaseRepository[User]):\n def __init__(self, conn: AsyncConnection):\n super().__init__(\n db_connection=conn,\n table_name=\"users\",\n model_class=User,\n primary_key=\"id\"\n )\n\n# Usage\nasync with tm.transaction() as conn:\n repo = UserRepository(conn)\n user = await repo.get_by_id(user_id)\n```\n\n### JSONB Support\n\n```python\nfrom typing import Dict, List, Any\nfrom pydantic import BaseModel\nfrom psycopg_toolkit import BaseRepository\n\nclass UserProfile(BaseModel):\n id: int\n name: str\n # These fields are automatically detected as JSONB\n metadata: Dict[str, Any]\n preferences: Dict[str, str]\n tags: List[str]\n\nclass UserRepository(BaseRepository[UserProfile, int]):\n def __init__(self, conn):\n super().__init__(\n db_connection=conn,\n table_name=\"user_profiles\",\n model_class=UserProfile,\n primary_key=\"id\"\n # auto_detect_json=True by default\n )\n\n# Usage - JSON fields handled automatically\nuser = UserProfile(\n id=1,\n name=\"John Doe\",\n metadata={\"created_at\": \"2024-01-01\", \"source\": \"web\"},\n preferences={\"theme\": \"dark\", \"language\": \"en\"},\n tags=[\"premium\", \"beta_tester\"]\n)\n\n# JSONB fields automatically serialized/deserialized\ncreated_user = await repo.create(user)\nretrieved_user = await repo.get_by_id(1)\n```\n\n### Schema Management\n\n```python\nfrom psycopg_toolkit.core.transaction import SchemaManager\n\nclass UserSchemaManager(SchemaManager[None]):\n async def create_schema(self, conn: AsyncConnection) -> None:\n await conn.execute(\"\"\"\n CREATE TABLE IF NOT EXISTS users (\n id UUID PRIMARY KEY,\n email TEXT UNIQUE NOT NULL\n )\n \"\"\")\n\n async def drop_schema(self, conn: AsyncConnection) -> None:\n await conn.execute(\"DROP TABLE IF EXISTS users\")\n\n# Usage\nasync with tm.with_schema(UserSchemaManager()) as _:\n # Schema available here\n pass # Automatically dropped after\n```\n\n## Error Handling\n\n```python\nfrom psycopg_toolkit import (\n DatabaseConnectionError,\n DatabasePoolError,\n DatabaseNotAvailable,\n RecordNotFoundError\n)\n\ntry:\n async with tm.transaction() as conn:\n repo = UserRepository(conn)\n user = await repo.get_by_id(user_id)\nexcept DatabaseConnectionError as e:\n print(f\"Connection error: {e.original_error}\")\nexcept RecordNotFoundError:\n print(f\"User {user_id} not found\")\n```\n\n## Documentation\n\n- [Database Management](docs/database.md)\n- [Transaction Management](docs/transaction_manager.md)\n- [Base Repository](docs/base_repository.md)\n- [JSONB Support](docs/jsonb_support.md)\n- [PsycopgHelper](docs/psycopg_helper.md)\n\n## Running Tests\n\n```bash\n# Install dependencies\nuv sync --all-groups\n\n# Run all tests except performance tests (default)\nuv run pytest\n\n# Run only performance tests\nuv run pytest -m performance\n\n# Run all tests including performance\nuv run pytest -m \"\"\n\n# Run specific test categories\nuv run pytest tests/unit/ # Only unit tests\nuv run pytest -m performance # Only performance tests\n\n# Run with coverage\nuv run pytest --cov=src/psycopg_toolkit --cov-report=html\n```\n\n### Test Categories\n\nThe test suite is organized into three categories:\n\n- **Unit tests**: Fast, isolated tests that don't require a database (in `tests/unit/`)\n- **Integration tests**: Tests that require a real PostgreSQL database (in `tests/` root)\n- **Performance tests**: Benchmarks and performance measurements (marked with `@pytest.mark.performance`)\n\nPerformance tests are excluded by default to keep the regular test runs fast. Use the `-m performance` flag to run them explicitly.\n\n## Contributing\n\n1. Fork the repository\n2. Create a feature branch\n3. Add tests for new features\n4. Ensure all tests pass\n5. Submit a pull request\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A Python PostgreSQL database utility with connection pooling",
"version": "0.2.0",
"project_urls": {
"Documentation": "https://github.com/descoped/psycopg-toolkit/tree/master/docs",
"Homepage": "https://github.com/descoped/psycopg-toolkit",
"Issues": "https://github.com/descoped/psycopg-toolkit/issues",
"Repository": "https://github.com/descoped/psycopg-toolkit"
},
"split_keywords": [
"async",
" database",
" jsonb",
" pool",
" postgresql",
" psycopg",
" repository-pattern"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "0baacbae4d90ac08069b8abbae7b19d011edb00c7fb7535c4f9c5d68179d8759",
"md5": "78911624813c50c95df7b64ebe99315a",
"sha256": "9e3afd9a4ce81d2a71901bb29c66e0c3b589ff1c3d4ab57a6b4d9c04326b5e74"
},
"downloads": -1,
"filename": "psycopg_toolkit-0.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "78911624813c50c95df7b64ebe99315a",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 23192,
"upload_time": "2025-08-05T12:38:06",
"upload_time_iso_8601": "2025-08-05T12:38:06.983448Z",
"url": "https://files.pythonhosted.org/packages/0b/aa/cbae4d90ac08069b8abbae7b19d011edb00c7fb7535c4f9c5d68179d8759/psycopg_toolkit-0.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "785d0b921f3b1624ee5f3eb37516c12406f4a7f0f2f4d5195908b1bffe8acd7f",
"md5": "6c13705e560596690ddcfaec6be73ebf",
"sha256": "8f398f5a026ba7c1efa161a425a37e80b15d21a1e0b6207e01dc9727a2c2d379"
},
"downloads": -1,
"filename": "psycopg_toolkit-0.2.0.tar.gz",
"has_sig": false,
"md5_digest": "6c13705e560596690ddcfaec6be73ebf",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 20561,
"upload_time": "2025-08-05T12:38:08",
"upload_time_iso_8601": "2025-08-05T12:38:08.167496Z",
"url": "https://files.pythonhosted.org/packages/78/5d/0b921f3b1624ee5f3eb37516c12406f4a7f0f2f4d5195908b1bffe8acd7f/psycopg_toolkit-0.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-05 12:38:08",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "descoped",
"github_project": "psycopg-toolkit",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "psycopg-toolkit"
}
JSON
