psycopg-toolkit


Namepsycopg-toolkit JSON
Version 0.1.7 PyPI version JSON
download
home_pagehttps://github.com/descoped/psycopg-toolkit
SummaryA Python PostgreSQL database utility with connection pooling
upload_time2025-02-08 16:20:24
maintainerNone
docs_urlNone
authorOve Ranheim
requires_python<4.0,>=3.11
licenseMIT
keywords postgresql database async pool psycopg
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # Psycopg Toolkit

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
- 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)
```

### 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](https://github.com/descoped/psycopg-toolkit/blob/master/docs/database.md)
- [Transaction Management](https://github.com/descoped/psycopg-toolkit/blob/master/docs/transaction_manager.md)
- [Base Repository](https://github.com/descoped/psycopg-toolkit/blob/master/docs/base_repository.md)
- [PsycopgHelper](https://github.com/descoped/psycopg-toolkit/blob/master/docs/psycopg_helper.md)

## Running Tests

```bash
# Install test dependencies
poetry install --with test

# Run tests
poetry run pytest
```

## 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": "https://github.com/descoped/psycopg-toolkit",
    "name": "psycopg-toolkit",
    "maintainer": null,
    "docs_url": null,
    "requires_python": "<4.0,>=3.11",
    "maintainer_email": null,
    "keywords": "postgresql, database, async, pool, psycopg",
    "author": "Ove Ranheim",
    "author_email": "oranheim@gmail.com",
    "download_url": "https://files.pythonhosted.org/packages/ea/69/4e8386152f1dee4b7db2d5988e5b20151265a4970b1e7250317f6d05fadb/psycopg_toolkit-0.1.7.tar.gz",
    "platform": null,
    "description": "# Psycopg Toolkit\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- 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### 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](https://github.com/descoped/psycopg-toolkit/blob/master/docs/database.md)\n- [Transaction Management](https://github.com/descoped/psycopg-toolkit/blob/master/docs/transaction_manager.md)\n- [Base Repository](https://github.com/descoped/psycopg-toolkit/blob/master/docs/base_repository.md)\n- [PsycopgHelper](https://github.com/descoped/psycopg-toolkit/blob/master/docs/psycopg_helper.md)\n\n## Running Tests\n\n```bash\n# Install test dependencies\npoetry install --with test\n\n# Run tests\npoetry run pytest\n```\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.1.7",
    "project_urls": {
        "Documentation": "https://github.com/descoped/psycopg-toolkit/tree/master/docs",
        "Homepage": "https://github.com/descoped/psycopg-toolkit",
        "Repository": "https://github.com/descoped/psycopg-toolkit"
    },
    "split_keywords": [
        "postgresql",
        " database",
        " async",
        " pool",
        " psycopg"
    ],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "320b7765e25663cc86490d779695879958e3dcdd3868eab6b7f0c3ff71ef74cc",
                "md5": "d697e3a03a1a5167fc4d7cb525d1226b",
                "sha256": "5e44b73d1026e6d1bc2d1246dc79fc3eb43802462df8f800b2c25b247e5740f8"
            },
            "downloads": -1,
            "filename": "psycopg_toolkit-0.1.7-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "d697e3a03a1a5167fc4d7cb525d1226b",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": "<4.0,>=3.11",
            "size": 15453,
            "upload_time": "2025-02-08T16:20:22",
            "upload_time_iso_8601": "2025-02-08T16:20:22.549465Z",
            "url": "https://files.pythonhosted.org/packages/32/0b/7765e25663cc86490d779695879958e3dcdd3868eab6b7f0c3ff71ef74cc/psycopg_toolkit-0.1.7-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "ea694e8386152f1dee4b7db2d5988e5b20151265a4970b1e7250317f6d05fadb",
                "md5": "497dc8d063c2130965d6131d1e529fd9",
                "sha256": "37cb05adc40c9cd7b4772b01884a8fc19a550db4a7371d2bb9ac53d2f2f330d2"
            },
            "downloads": -1,
            "filename": "psycopg_toolkit-0.1.7.tar.gz",
            "has_sig": false,
            "md5_digest": "497dc8d063c2130965d6131d1e529fd9",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": "<4.0,>=3.11",
            "size": 12927,
            "upload_time": "2025-02-08T16:20:24",
            "upload_time_iso_8601": "2025-02-08T16:20:24.405237Z",
            "url": "https://files.pythonhosted.org/packages/ea/69/4e8386152f1dee4b7db2d5988e5b20151265a4970b1e7250317f6d05fadb/psycopg_toolkit-0.1.7.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-02-08 16:20:24",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "descoped",
    "github_project": "psycopg-toolkit",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": false,
    "lcname": "psycopg-toolkit"
}
        
Elapsed time: 1.24013s