| Name | pgdbm JSON |
| Version |
0.1.0
JSON |
| download |
| home_page | None |
| Summary | Production-ready async PostgreSQL utilities with connection pooling, migrations, and testing support |
| upload_time | 2025-08-13 19:52:13 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.9 |
| license | MIT |
| keywords |
async
asyncio
asyncpg
database
migrations
postgresql
testing
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# pgdbm
[](https://opensource.org/licenses/MIT)
[](https://pypi.org/project/pgdbm/)
[](https://pypi.org/project/pgdbm/)
A PostgreSQL library for Python that provides high-level async database operations with built-in migration management, connection pooling, and testing utilities. It offers:
- **Connection pooling** with proper resource management
- **Schema migrations** that are version-controlled and automated
- **Testing utilities** that provide isolated test databases
- **Module isolation** when multiple services share a database
- **Monitoring** for slow queries and connection issues
## Key Features
- **🚀 High Performance** - Built on asyncpg, the fastest PostgreSQL driver for Python
- **📦 Migration System** - Version-controlled schema migrations with automatic ordering
- **🧪 Testing Support** - Fixtures and utilities for database testing
- **🔧 Module Isolation** - Prevent table conflicts when modules share databases
- **📊 Monitoring** - Track slow queries and connection pool metrics
- **🔒 Type Safe** - Full type hints and Pydantic integration
### Design intent: dual ownership
pgdbm is designed so a module can either:
- **Own its database** when run independently (it creates a pool and runs its own migrations), or
- **Use a database owned by another** component via a shared pool while still running its own migrations into a schema namespace.
This keeps modules portable: the same code can run as a standalone service or as part of a larger app.
## Installation
```bash
# Install using uv (recommended)
uv add pgdbm
# Or using pip
pip install pgdbm
```
## Quick Start
### 1. Create a migration file
```sql
-- migrations/001_initial.sql
CREATE TABLE IF NOT EXISTS {{tables.users}} (
id SERIAL PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
```
### 2. Use pgdbm
```python
from pgdbm import AsyncDatabaseManager, DatabaseConfig, AsyncMigrationManager
# Configure and connect
config = DatabaseConfig(
host="localhost",
database="myapp",
user="postgres",
password="secret"
)
db = AsyncDatabaseManager(config)
await db.connect()
# Apply migrations
migrations = AsyncMigrationManager(db, migrations_path="./migrations")
await migrations.apply_pending()
# Use your database
user_id = await db.execute_and_return_id(
"INSERT INTO {{tables.users}} (email) VALUES ($1)",
"user@example.com"
)
# Clean up
await db.disconnect()
```
Standalone vs shared-pool usage:
```python
# Standalone (owns the DB):
config = DatabaseConfig(connection_string="postgresql://localhost/app", schema="users")
db = AsyncDatabaseManager(config)
await db.connect()
await AsyncMigrationManager(db, "./migrations", module_name="users").apply_pending()
# Shared pool (uses external DB):
shared = await AsyncDatabaseManager.create_shared_pool(DatabaseConfig(connection_string="postgresql://localhost/app"))
db = AsyncDatabaseManager(pool=shared, schema="users") # Module still runs its own migrations
await AsyncMigrationManager(db, "./migrations", module_name="users").apply_pending()
```
## Core Patterns
### 1. Migration Management
pgdbm includes a built-in migration system:
```python
# Apply all pending migrations
migrations = AsyncMigrationManager(db, migrations_path="./migrations")
result = await migrations.apply_pending()
# Check what was applied
for migration in result['applied']:
print(f"Applied {migration['filename']} in {migration['execution_time_ms']}ms")
```
Migrations are automatically ordered by version number extracted from filenames.
### 2. Module Isolation
When multiple modules share a database, use schemas to prevent table conflicts:
```python
# Each module can get its own schema
user_db = AsyncDatabaseManager(
config=DatabaseConfig(database="app", schema="user_module")
)
blog_db = AsyncDatabaseManager(
config=DatabaseConfig(database="app", schema="blog_module")
)
# Both can have a "users" table without conflict:
# - user_module.users
# - blog_module.users
```
The `{{tables.tablename}}` syntax in queries automatically expands to the correct schema-qualified name.
### 3. Connection Pool Sharing
For applications with multiple services sharing a database:
```python
# Create shared pool
shared_pool = await AsyncDatabaseManager.create_shared_pool(config)
# Each service gets its own schema but shares connections
user_db = AsyncDatabaseManager(pool=shared_pool, schema="users")
billing_db = AsyncDatabaseManager(pool=shared_pool, schema="billing")
```
### 4. Testing Support
Built-in fixtures for database tests:
```python
# conftest.py
from pgdbm.fixtures.conftest import *
# test_users.py
async def test_create_user(test_db):
# Automatic test database with cleanup
await test_db.execute("""
CREATE TABLE users (id SERIAL PRIMARY KEY, email TEXT)
""")
user_id = await test_db.execute_and_return_id(
"INSERT INTO users (email) VALUES ($1)",
"test@example.com"
)
assert user_id == 1
```
### 5. Monitoring
Track database performance:
```python
from pgdbm import MonitoredAsyncDatabaseManager
db = MonitoredAsyncDatabaseManager(
config=config,
slow_query_threshold_ms=100 # Log queries over 100ms
)
# Get metrics
metrics = await db.get_query_metrics()
slow_queries = await db.get_slow_queries(limit=10)
```
### 6. Production TLS and Timeouts
Enable TLS with certificate verification and enforce server-side timeouts:
```python
from pgdbm import AsyncDatabaseManager, DatabaseConfig
config = DatabaseConfig(
connection_string="postgresql://db.example.com/app",
ssl_enabled=True,
ssl_mode="verify-full", # 'require' | 'verify-ca' | 'verify-full'
ssl_ca_file="/etc/ssl/certs/ca.pem",
# Optional mutual TLS:
# ssl_cert_file="/etc/ssl/certs/client.crt",
# ssl_key_file="/etc/ssl/private/client.key",
# Sensible timeouts (ms)
statement_timeout_ms=60_000,
idle_in_transaction_session_timeout_ms=60_000,
lock_timeout_ms=5_000,
)
db = AsyncDatabaseManager(config)
await db.connect()
```
Notes:
- Use `verify-full` for strict hostname and certificate validation in production.
- Timeouts are applied via `server_settings`; you can override or disable by passing None.
## Examples
The `examples/` directory contains applications:
- **todo-app/** - REST API with migrations, testing, and error handling
- **saas-app/** - Multi-tenant SaaS application
- **microservices/** - Multiple services sharing a connection pool
## Documentation
- [Quickstart Guide](docs/quickstart.md) - Get started
- [Patterns Guide](docs/patterns.md) - Deployment patterns, schema isolation, and framework integration
- [Migration Guide](docs/migrations.md) - Schema versioning and {{tables.}} syntax
- [API Reference](docs/api-reference.md) - Complete API documentation
- [Testing Guide](docs/testing.md) - Testing best practices
## Contributing
Short version:
- Requirements: Python 3.9+, PostgreSQL 12+, uv (or pip)
- Setup:
- uv: `uv sync`
- pip: `pip install -e ".[dev]"`
- hooks: `pre-commit install`
- Run tests: `pytest`
- Lint/type-check: `pre-commit run --all-files`
Notes:
- Integration tests use ephemeral databases; you can override with env vars like `TEST_DB_HOST`, `TEST_DB_PORT`, `TEST_DB_USER`, `TEST_DB_PASSWORD`.
- Keep PRs small and focused, include tests/docs for user-visible changes.
- Style is enforced via Black/Isort/Ruff/Mypy; run pre-commit locally before pushing.
## License
MIT License - see [LICENSE](LICENSE) for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "pgdbm",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": "Juan Reyero <juan@juanreyero.com>",
"keywords": "async, asyncio, asyncpg, database, migrations, postgresql, testing",
"author": null,
"author_email": "Juan Reyero <juan@juanreyero.com>",
"download_url": "https://files.pythonhosted.org/packages/3b/39/dad7ba86632ccd8fdfd92e1cf0e3f209ee36e8e4eafcba4fa7ec9f24af36/pgdbm-0.1.0.tar.gz",
"platform": null,
"description": "# pgdbm\n\n[](https://opensource.org/licenses/MIT)\n[](https://pypi.org/project/pgdbm/)\n[](https://pypi.org/project/pgdbm/)\n\nA PostgreSQL library for Python that provides high-level async database operations with built-in migration management, connection pooling, and testing utilities. It offers:\n\n- **Connection pooling** with proper resource management\n- **Schema migrations** that are version-controlled and automated\n- **Testing utilities** that provide isolated test databases\n- **Module isolation** when multiple services share a database\n- **Monitoring** for slow queries and connection issues\n\n## Key Features\n\n- **\ud83d\ude80 High Performance** - Built on asyncpg, the fastest PostgreSQL driver for Python\n- **\ud83d\udce6 Migration System** - Version-controlled schema migrations with automatic ordering\n- **\ud83e\uddea Testing Support** - Fixtures and utilities for database testing\n- **\ud83d\udd27 Module Isolation** - Prevent table conflicts when modules share databases\n- **\ud83d\udcca Monitoring** - Track slow queries and connection pool metrics\n- **\ud83d\udd12 Type Safe** - Full type hints and Pydantic integration\n\n### Design intent: dual ownership\n\npgdbm is designed so a module can either:\n- **Own its database** when run independently (it creates a pool and runs its own migrations), or\n- **Use a database owned by another** component via a shared pool while still running its own migrations into a schema namespace.\n\nThis keeps modules portable: the same code can run as a standalone service or as part of a larger app.\n\n## Installation\n\n```bash\n# Install using uv (recommended)\nuv add pgdbm\n\n# Or using pip\npip install pgdbm\n```\n\n## Quick Start\n\n### 1. Create a migration file\n\n```sql\n-- migrations/001_initial.sql\nCREATE TABLE IF NOT EXISTS {{tables.users}} (\n id SERIAL PRIMARY KEY,\n email VARCHAR(255) UNIQUE NOT NULL,\n created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP\n);\n```\n\n### 2. Use pgdbm\n\n```python\nfrom pgdbm import AsyncDatabaseManager, DatabaseConfig, AsyncMigrationManager\n\n# Configure and connect\nconfig = DatabaseConfig(\n host=\"localhost\",\n database=\"myapp\",\n user=\"postgres\",\n password=\"secret\"\n)\n\ndb = AsyncDatabaseManager(config)\nawait db.connect()\n\n# Apply migrations\nmigrations = AsyncMigrationManager(db, migrations_path=\"./migrations\")\nawait migrations.apply_pending()\n\n# Use your database\nuser_id = await db.execute_and_return_id(\n \"INSERT INTO {{tables.users}} (email) VALUES ($1)\",\n \"user@example.com\"\n)\n\n# Clean up\nawait db.disconnect()\n```\n\nStandalone vs shared-pool usage:\n\n```python\n# Standalone (owns the DB):\nconfig = DatabaseConfig(connection_string=\"postgresql://localhost/app\", schema=\"users\")\ndb = AsyncDatabaseManager(config)\nawait db.connect()\nawait AsyncMigrationManager(db, \"./migrations\", module_name=\"users\").apply_pending()\n\n# Shared pool (uses external DB):\nshared = await AsyncDatabaseManager.create_shared_pool(DatabaseConfig(connection_string=\"postgresql://localhost/app\"))\ndb = AsyncDatabaseManager(pool=shared, schema=\"users\") # Module still runs its own migrations\nawait AsyncMigrationManager(db, \"./migrations\", module_name=\"users\").apply_pending()\n```\n\n## Core Patterns\n\n### 1. Migration Management\n\npgdbm includes a built-in migration system:\n\n```python\n# Apply all pending migrations\nmigrations = AsyncMigrationManager(db, migrations_path=\"./migrations\")\nresult = await migrations.apply_pending()\n\n# Check what was applied\nfor migration in result['applied']:\n print(f\"Applied {migration['filename']} in {migration['execution_time_ms']}ms\")\n```\n\nMigrations are automatically ordered by version number extracted from filenames.\n\n### 2. Module Isolation\n\nWhen multiple modules share a database, use schemas to prevent table conflicts:\n\n```python\n# Each module can get its own schema\nuser_db = AsyncDatabaseManager(\n config=DatabaseConfig(database=\"app\", schema=\"user_module\")\n)\nblog_db = AsyncDatabaseManager(\n config=DatabaseConfig(database=\"app\", schema=\"blog_module\")\n)\n\n# Both can have a \"users\" table without conflict:\n# - user_module.users\n# - blog_module.users\n```\n\nThe `{{tables.tablename}}` syntax in queries automatically expands to the correct schema-qualified name.\n\n### 3. Connection Pool Sharing\n\nFor applications with multiple services sharing a database:\n\n```python\n# Create shared pool\nshared_pool = await AsyncDatabaseManager.create_shared_pool(config)\n\n# Each service gets its own schema but shares connections\nuser_db = AsyncDatabaseManager(pool=shared_pool, schema=\"users\")\nbilling_db = AsyncDatabaseManager(pool=shared_pool, schema=\"billing\")\n```\n\n### 4. Testing Support\n\nBuilt-in fixtures for database tests:\n\n```python\n# conftest.py\nfrom pgdbm.fixtures.conftest import *\n\n# test_users.py\nasync def test_create_user(test_db):\n # Automatic test database with cleanup\n await test_db.execute(\"\"\"\n CREATE TABLE users (id SERIAL PRIMARY KEY, email TEXT)\n \"\"\")\n\n user_id = await test_db.execute_and_return_id(\n \"INSERT INTO users (email) VALUES ($1)\",\n \"test@example.com\"\n )\n assert user_id == 1\n```\n\n### 5. Monitoring\n\nTrack database performance:\n\n```python\nfrom pgdbm import MonitoredAsyncDatabaseManager\n\ndb = MonitoredAsyncDatabaseManager(\n config=config,\n slow_query_threshold_ms=100 # Log queries over 100ms\n)\n\n# Get metrics\nmetrics = await db.get_query_metrics()\nslow_queries = await db.get_slow_queries(limit=10)\n```\n\n### 6. Production TLS and Timeouts\n\nEnable TLS with certificate verification and enforce server-side timeouts:\n\n```python\nfrom pgdbm import AsyncDatabaseManager, DatabaseConfig\n\nconfig = DatabaseConfig(\n connection_string=\"postgresql://db.example.com/app\",\n ssl_enabled=True,\n ssl_mode=\"verify-full\", # 'require' | 'verify-ca' | 'verify-full'\n ssl_ca_file=\"/etc/ssl/certs/ca.pem\",\n # Optional mutual TLS:\n # ssl_cert_file=\"/etc/ssl/certs/client.crt\",\n # ssl_key_file=\"/etc/ssl/private/client.key\",\n\n # Sensible timeouts (ms)\n statement_timeout_ms=60_000,\n idle_in_transaction_session_timeout_ms=60_000,\n lock_timeout_ms=5_000,\n)\n\ndb = AsyncDatabaseManager(config)\nawait db.connect()\n```\n\nNotes:\n- Use `verify-full` for strict hostname and certificate validation in production.\n- Timeouts are applied via `server_settings`; you can override or disable by passing None.\n\n## Examples\n\nThe `examples/` directory contains applications:\n\n- **todo-app/** - REST API with migrations, testing, and error handling\n- **saas-app/** - Multi-tenant SaaS application\n- **microservices/** - Multiple services sharing a connection pool\n\n## Documentation\n\n- [Quickstart Guide](docs/quickstart.md) - Get started\n- [Patterns Guide](docs/patterns.md) - Deployment patterns, schema isolation, and framework integration\n- [Migration Guide](docs/migrations.md) - Schema versioning and {{tables.}} syntax\n- [API Reference](docs/api-reference.md) - Complete API documentation\n- [Testing Guide](docs/testing.md) - Testing best practices\n\n## Contributing\n\nShort version:\n\n- Requirements: Python 3.9+, PostgreSQL 12+, uv (or pip)\n- Setup:\n - uv: `uv sync`\n - pip: `pip install -e \".[dev]\"`\n - hooks: `pre-commit install`\n- Run tests: `pytest`\n- Lint/type-check: `pre-commit run --all-files`\n\nNotes:\n\n- Integration tests use ephemeral databases; you can override with env vars like `TEST_DB_HOST`, `TEST_DB_PORT`, `TEST_DB_USER`, `TEST_DB_PASSWORD`.\n- Keep PRs small and focused, include tests/docs for user-visible changes.\n- Style is enforced via Black/Isort/Ruff/Mypy; run pre-commit locally before pushing.\n\n## License\n\nMIT License - see [LICENSE](LICENSE) for details.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Production-ready async PostgreSQL utilities with connection pooling, migrations, and testing support",
"version": "0.1.0",
"project_urls": {
"Changelog": "https://github.com/juanre/pgdbm/blob/main/CHANGELOG.md",
"Documentation": "https://github.com/juanre/pgdbm",
"Homepage": "https://juanreyero.com/open/pgdbm",
"Issues": "https://github.com/juanre/pgdbm/issues",
"Repository": "https://github.com/juanre/pgdbm"
},
"split_keywords": [
"async",
" asyncio",
" asyncpg",
" database",
" migrations",
" postgresql",
" testing"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "00830e38f646b4869ddc0120f9b65785fded31644466fd7dde9aa7cd66bf0b75",
"md5": "2583a820b06e82f62f4abf35b9e16127",
"sha256": "b8eacee31763d8dab9311b9dfb8edfc79413473dffc7e50a10b3d7546ffe0ad6"
},
"downloads": -1,
"filename": "pgdbm-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2583a820b06e82f62f4abf35b9e16127",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 30375,
"upload_time": "2025-08-13T19:52:11",
"upload_time_iso_8601": "2025-08-13T19:52:11.194882Z",
"url": "https://files.pythonhosted.org/packages/00/83/0e38f646b4869ddc0120f9b65785fded31644466fd7dde9aa7cd66bf0b75/pgdbm-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "3b39dad7ba86632ccd8fdfd92e1cf0e3f209ee36e8e4eafcba4fa7ec9f24af36",
"md5": "eb73f2db03de67eb1afbdd80876a4026",
"sha256": "4ba0d8137bbf98b66f446d7a3b279feb86e3038f77c0496059e2e973c7516144"
},
"downloads": -1,
"filename": "pgdbm-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "eb73f2db03de67eb1afbdd80876a4026",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 191693,
"upload_time": "2025-08-13T19:52:13",
"upload_time_iso_8601": "2025-08-13T19:52:13.066066Z",
"url": "https://files.pythonhosted.org/packages/3b/39/dad7ba86632ccd8fdfd92e1cf0e3f209ee36e8e4eafcba4fa7ec9f24af36/pgdbm-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-13 19:52:13",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "juanre",
"github_project": "pgdbm",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "pgdbm"
}
