netrun-db-pool 2.0.0
Production-grade async database connection pooling for Netrun Systems services
License: MIT
Requires Python: >=3.10
Keywords: asyncpg, connection-pool, database, fastapi, multi-tenant, namespace, netrun, postgresql, rls, sqlalchemy

# netrun-db-pool

Production-grade async database connection pooling for Netrun Systems services.

## IMPORTANT: Migration to v2.0.0 (Namespace Structure)

**Version 2.0.0 introduces a breaking change in import paths.** The package has been migrated to the `netrun.db` namespace structure for consistency with other Netrun packages.

### New Import Path

```python
# Old (v1.x - deprecated but still works with warnings)
from netrun_db_pool import AsyncDatabasePool, PoolConfig
from netrun_db_pool.middleware import get_db_dependency

# New (v2.0+ - recommended)
from netrun.db import AsyncDatabasePool, PoolConfig
from netrun.db.middleware import get_db_dependency
```

### Backwards Compatibility

Version 2.0.0 maintains full backwards compatibility through deprecation shims. Old imports will continue to work but will issue `DeprecationWarning`. The old import path will be removed in version 3.0.0 (estimated Q2 2026).

### Migration Checklist

- [ ] Update all imports from `netrun_db_pool` to `netrun.db`
- [ ] Update all imports from `netrun_db_pool.middleware` to `netrun.db.middleware`
- [ ] Update `requirements.txt` or `pyproject.toml` to use `netrun-db-pool>=2.0.0`
- [ ] Test thoroughly in development environment
- [ ] Deploy to production

## Features

- **AsyncPG Performance**: SQLAlchemy 2.0+ with asyncpg driver for maximum PostgreSQL performance
- **Multi-Tenant RLS**: Built-in Row-Level Security support for SaaS applications
- **Health Monitoring**: Comprehensive pool metrics and health checks
- **FastAPI Integration**: Drop-in dependency injection and middleware
- **Production Ready**: Connection pre-ping, automatic recycling, configurable pooling
- **Type Safe**: Full Pydantic configuration with environment variable support

## Installation

```bash
pip install netrun-db-pool
```

### Optional Dependencies

```bash
# FastAPI integration
pip install netrun-db-pool[fastapi]

# Development tools
pip install netrun-db-pool[dev]
```

## Quick Start

### Basic Usage

```python
from netrun.db import AsyncDatabasePool, PoolConfig
from sqlalchemy import select, text

# Create pool from environment variables
pool = AsyncDatabasePool.from_env()
await pool.initialize()

# Use with context manager
async with pool.get_session() as session:
    result = await session.execute(text("SELECT version()"))
    print(result.scalar())
```

### FastAPI Integration

```python
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from netrun.db import AsyncDatabasePool
from netrun.db.middleware import get_db_dependency

app = FastAPI()
pool = AsyncDatabasePool.from_env()

# Create dependency
get_db = get_db_dependency(pool)

@app.on_event("startup")
async def startup():
    await pool.initialize()

@app.on_event("shutdown")
async def shutdown():
    await pool.close()

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

@app.get("/health/db")
async def db_health():
    health = await pool.health_check_detailed()
    return health.to_dict()
```

### Multi-Tenant with RLS

```python
from fastapi import FastAPI, Request, Depends
from netrun.db import TenantAwareDatabasePool
from netrun.db.middleware import (
    get_tenant_db_dependency,
    TenantContextMiddleware,
)

app = FastAPI()
pool = TenantAwareDatabasePool.from_env()

# Extract tenant_id from JWT
def extract_tenant_id(request: Request) -> str:
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    claims = decode_jwt(token)  # Your JWT decoder
    return claims.get("tenant_id")

def get_tenant_id(request: Request) -> str:
    return request.state.tenant_id

# Add middleware to extract tenant context
app.add_middleware(
    TenantContextMiddleware,
    tenant_id_extractor=extract_tenant_id,
)

# Create tenant-aware dependency
get_tenant_db = get_tenant_db_dependency(
    pool=pool,
    tenant_id_extractor=get_tenant_id,
)

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_tenant_db)):
    # Only current tenant's users are accessible (RLS enforced)
    result = await db.execute(select(User))
    return result.scalars().all()
```

### PostgreSQL RLS Setup

```sql
-- Enable RLS on table
ALTER TABLE users ENABLE ROW LEVEL SECURITY;

-- Create RLS policy
CREATE POLICY tenant_isolation ON users
    USING (tenant_id = current_setting('app.current_tenant_id')::uuid);

-- Verify RLS is enabled
SELECT relname, relrowsecurity
FROM pg_class
WHERE relname = 'users';
```

## Configuration

### Environment Variables

```bash
# Database connection
DATABASE_URL=postgresql+asyncpg://user:pass@localhost/db

# Pool settings
DB_POOL_SIZE=20                # Base pool size
DB_MAX_OVERFLOW=10             # Overflow connections (total max: 30)
DB_POOL_TIMEOUT=30             # Connection acquisition timeout (seconds)
DB_POOL_RECYCLE=3600           # Recycle connections after 1 hour
DB_COMMAND_TIMEOUT=60          # Query timeout (seconds)

# Application settings
DB_APP_NAME=my-service         # PostgreSQL application name
DB_ENABLE_JIT=false            # Disable JIT for consistent performance
DB_ECHO=false                  # Enable SQLAlchemy query logging (dev only)
```

### Programmatic Configuration

```python
from netrun.db import AsyncDatabasePool, PoolConfig

config = PoolConfig(
    database_url="postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_recycle=3600,
    app_name="my-service",
)

pool = AsyncDatabasePool(config=config)
```

## Health Monitoring

### Simple Health Check

```python
is_healthy = await pool.health_check()
print(f"Database healthy: {is_healthy}")
```

### Detailed Health Metrics

```python
health = await pool.health_check_detailed()
print(f"Status: {health.status}")
print(f"Pool utilization: {health.utilization_percent}%")
print(f"Available connections: {health.available_connections}")
print(f"Checked out: {health.checked_out}")
print(f"Latency: {health.latency_ms}ms")

# Convert to dict for JSON response
return health.to_dict()
```

### Health Check Response

```json
{
  "healthy": true,
  "status": "healthy",
  "pool_size": 20,
  "checked_in_connections": 18,
  "checked_out_connections": 2,
  "overflow_connections": 0,
  "max_overflow": 10,
  "utilization_percent": 6.67,
  "available_connections": 28,
  "latency_ms": 2.34,
  "database": "postgresql",
  "timestamp": "2025-11-25T10:30:00Z"
}
```

## Advanced Features

### Verify RLS is Enabled

```python
pool = TenantAwareDatabasePool.from_env()

# Check if RLS is enabled for a table
is_protected = await pool.verify_rls_enabled("users")
if not is_protected:
    logger.warning("RLS not enabled for users table!")
```

### Get Current Tenant Context

```python
async with pool.get_tenant_session(tenant_id="550e8400-...") as session:
    current_tenant = await pool.get_current_tenant_context(session)
    print(f"Operating in tenant context: {current_tenant}")
```

### Custom Session Factory

```python
# Access SQLAlchemy engine for advanced use cases
engine = pool.engine

# Create custom session factory
from sqlalchemy.ext.asyncio import async_sessionmaker

custom_factory = async_sessionmaker(
    engine,
    expire_on_commit=True,  # Different from default
    autoflush=True,
)
```

## Best Practices

### Connection Pool Sizing

```python
# For typical web applications (FastAPI, Django)
pool_size = min(
    10 * num_cpu_cores,  # 10 connections per core
    max_database_connections / num_app_instances,
)

# Example: 4 cores, 100 max DB connections, 2 app instances
# pool_size = min(40, 50) = 40
# max_overflow = pool_size * 0.5 = 20
```

### PostgreSQL Optimizations

```python
config = PoolConfig(
    database_url="postgresql+asyncpg://...",
    pool_pre_ping=True,      # Verify connections before use
    pool_recycle=3600,       # Recycle hourly (Azure requires < 4 hours)
    enable_jit=False,        # Disable JIT for consistent query planning
    command_timeout=60,      # Prevent long-running queries
)
```

### Error Handling

```python
from sqlalchemy.exc import OperationalError, TimeoutError

async with pool.get_session() as session:
    try:
        result = await session.execute(select(User))
        return result.scalars().all()
    except OperationalError as e:
        logger.error(f"Database connection error: {e}")
        raise
    except TimeoutError:
        logger.error("Query timeout exceeded")
        raise
```

## Testing

### Unit Tests with SQLite

```python
import pytest
from netrun.db import AsyncDatabasePool, PoolConfig

@pytest.fixture
async def db_pool():
    config = PoolConfig(
        database_url="sqlite+aiosqlite:///:memory:",
        pool_size=5,
        echo=True,
    )
    pool = AsyncDatabasePool(config=config)
    await pool.initialize()
    yield pool
    await pool.close()

@pytest.mark.asyncio
async def test_health_check(db_pool):
    health = await db_pool.health_check_detailed()
    assert health.healthy is True
    assert health.status == "healthy"
```

### Integration Tests with PostgreSQL

```bash
# docker-compose.test.yml
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: test_password
      POSTGRES_DB: test_db
    ports:
      - "5432:5432"
```

```python
@pytest.fixture(scope="session")
async def db_pool():
    config = PoolConfig(
        database_url="postgresql+asyncpg://postgres:test_password@localhost/test_db",
    )
    pool = AsyncDatabasePool(config=config)
    await pool.initialize()
    yield pool
    await pool.close()
```

## Performance Benchmarks

Based on Netrun Systems internal testing (PostgreSQL on Azure Flexible Server):

| Metric | AsyncDatabasePool | Native asyncpg | psycopg3 |
|--------|------------------|----------------|----------|
| Query Latency (p95) | 8.2ms | 7.1ms | 12.4ms |
| Throughput (queries/sec) | 1,240 | 1,380 | 890 |
| Connection Acquisition | 0.8ms | 0.5ms | 1.2ms |
| Memory per connection | 2.1MB | 1.8MB | 3.4MB |

**Conclusion**: AsyncDatabasePool provides 95% of native asyncpg performance while adding ORM benefits, health monitoring, and RLS support.

## Migration from Existing Pools

### From SQLAlchemy async_sessionmaker

```python
# Before
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(DATABASE_URL, pool_size=20)
AsyncSessionLocal = async_sessionmaker(engine)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# After
from netrun.db import AsyncDatabasePool
from netrun.db.middleware import get_db_dependency

pool = AsyncDatabasePool.from_env()
get_db = get_db_dependency(pool)
```

### From Native asyncpg

```python
# Before
import asyncpg

pool = await asyncpg.create_pool(
    host="localhost",
    database="mydb",
    user="user",
    password="pass",
    min_size=10,
    max_size=20,
)

async with pool.acquire() as conn:
    rows = await conn.fetch("SELECT * FROM users")

# After
from netrun.db import AsyncDatabasePool

pool = AsyncDatabasePool.from_env()
await pool.initialize()

async with pool.get_session() as session:
    result = await session.execute(select(User))
    users = result.scalars().all()
```

## Contributing

Contributions welcome! Please submit issues and pull requests to:
https://github.com/netrunsystems/netrun-db-pool

## License

MIT License - see LICENSE file for details.

## Support

- Documentation: https://github.com/netrunsystems/netrun-db-pool
- Issues: https://github.com/netrunsystems/netrun-db-pool/issues
- Email: engineering@netrunsystems.com

## Credits

Developed by Netrun Systems as part of the Service Consolidation Initiative (2025).
Based on production patterns from 8+ services across the Netrun portfolio.

**Related Projects**:
- Service #03 (Intirkast): Multi-tenant SaaS platform
- Service #71: Unified Database Pool (this project)
- Service #61: Unified Logging

---

**netrun-db-pool v2.0.0** - Production-grade database pooling for the async era.

**Breaking Changes in v2.0.0**: Migrated to `netrun.db` namespace. See migration guide above.