Structum Database (`structum-database`)¶

Structum Database offre un layer di astrazione robusto per SQLAlchemy (Sync/Async) e PostgreSQL (psycopg3).

Feature	Stato	Versione
Stato	Alpha	0.1.0
Namespace	`structum_lab.plugins.database`
Backend	SQLAlchemy, PostgreSQL

Indice¶

Cos’è Database Plugin
Concetti Fondamentali
Quick Start (5 Minuti)
Backend SQLAlchemy
Backend PostgreSQL
Connection Pooling
Transactions
Health Checks
Configuration
FastAPI Integration
Testing
Production Best Practices
API Reference

1. Cos’è Database Plugin¶

structum-database fornisce database connectivity production-ready con connection pooling, health monitoring, e transaction management.

1.1 Problema Risolto¶

Prima (senza plugin):

# ❌ Connection management manuale
import psycopg2

conn = psycopg2.connect("postgresql://...")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
# Nessun pooling
# Nessun health check
# Transaction management manuale

Dopo (con plugin):

# ✅ Connection pooling automatico
# ✅ Health checks integrati
# ✅ Transaction context manager
# ✅ Type-safe queries

db = SQLAlchemyDatabase.from_config()

with db.transaction() as conn:
    conn.execute("SELECT * FROM users WHERE id = :id", {"id": 1})
    user = conn.fetchone()

1.2 Caratteristiche Principali¶

Feature	Descrizione
Multiple Backends	SQLAlchemy (universal) o psycopg3 (PostgreSQL-only)
Connection Pooling	Automatic pool management con pre-ping
Transaction Management	Context managers per ACID guarantees
Health Checks	Built-in health monitoring con latency metrics
Type Safety	Full type hints per mypy strict
Config Integration	Setup via Dynaconf con secrets isolation

2. Concetti Fondamentali¶

2.1 Backend Comparison¶

Feature	SQLAlchemyDatabase	PostgresDatabase
Databases Supported	PostgreSQL, MySQL, SQLite, Oracle, MSSQL	PostgreSQL only
Performance	Good	Excellent (native driver)
ORM Support	✅ Full SQLAlchemy ORM	❌ Raw SQL only
Async Support	via asyncio extension	✅ Native
Pool Management	SQLAlchemy Engine	psycopg3 Pool
Dependencies	sqlalchemy>=2.0	psycopg[binary]>=3.1

Recommendation:

Use SQLAlchemyDatabase per progetti multi-database o con ORM
Use PostgresDatabase per maximum performance PostgreSQL-only

2.2 Connection Lifecycle¶

┌─────────────────────────────────────────┐
│ Application Startup                     │
├─────────────────────────────────────────┤
│ db = SQLAlchemyDatabase.from_config()  │
│   └─ Create connection pool             │
│      (min_size=2, max_size=10)          │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│ Request Handling                        │
├─────────────────────────────────────────┤
│ with db.transaction() as conn:         │
│   └─ Acquire connection from pool      │
│      Execute query                      │
│      Commit on success / Rollback       │
│      Release connection to pool         │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│ Application Shutdown                    │
├─────────────────────────────────────────┤
│ db.shutdown()                           │
│   └─ Close all pool connections        │
└─────────────────────────────────────────┘

3. Quick Start (5 Minuti)¶

Step 1: Installazione¶

# SQLAlchemy backend
pip install -e packages/database[sqlalchemy]

# PostgreSQL backend
pip install -e packages/database[postgres]

# Both
pip install -e packages/database[all]

Step 2: Configurazione¶

File: config/app/database.toml

[default]
url = "postgresql://user:password@localhost:5432/mydb"
pool_size = 10
pool_timeout = 30
echo = false  # Set true to log SQL queries

File: config/.secrets.toml

[database]
url = "postgresql://user:YOUR_PASSWORD@prod.db.example.com:5432/production_db"

Step 3: Basic Usage¶

from structum_lab.plugins.database import SQLAlchemyDatabase
from structum_lab.config import set_config_provider
from structum_lab.plugins.dynaconf import DynaconfConfigProvider

# Setup config
provider = DynaconfConfigProvider(root_path=".")
provider.auto_discover()
set_config_provider(provider)

# Create database connection
db = SQLAlchemyDatabase.from_config()

# Execute query in transaction
with db.transaction() as conn:
    result = conn.execute(
        "SELECT * FROM users WHERE id = :user_id",
        {"user_id": 1}
    )
    user = result.fetchone()
    
    if user:
        print(f"Found user: {user['username']}")

# Shutdown (cleanup)
db.shutdown()

4. Backend SQLAlchemy¶

4.1 Initialization¶

from structum_lab.plugins.database import SQLAlchemyDatabase

# From config
db = SQLAlchemyDatabase.from_config()

# Manual initialization
db = SQLAlchemyDatabase(
    url="postgresql://user:pass@localhost/mydb",
    pool_size=10,
    pool_timeout=30,
    pool_recycle=3600,  # Recycle connections after 1 hour
    echo=False,  # Log SQL queries
    pool_pre_ping=True  # Test connections before use
)

4.2 Query Execution¶

Select Query:

with db.transaction() as conn:
    result = conn.execute(
        "SELECT id, username, email FROM users WHERE active = :active",
        {"active": True}
    )
    
    # Fetch one
    user = result.fetchone()
    if user:
        print(f"{user['id']}: {user['username']}")
    
    # Fetch all
    users = result.fetchall()
    for user in users:
        print(f"{user['id']}: {user['username']}")
    
    # Fetch many
    batch = result.fetchmany(size=100)

Insert Query:

with db.transaction() as conn:
    result = conn.execute(
        """
        INSERT INTO users (username, email, created_at)
        VALUES (:username, :email, :created_at)
        RETURNING id
        """,
        {
            "username": "john_doe",
            "email": "john@example.com",
            "created_at": datetime.utcnow()
        }
    )
    
    new_id = result.fetchone()["id"]
    print(f"Created user with ID: {new_id}")

Update Query:

with db.transaction() as conn:
    result = conn.execute(
        """
        UPDATE users 
        SET last_login = :now
        WHERE id = :user_id
        """,
        {
            "now": datetime.utcnow(),
            "user_id": 123
        }
    )
    
    rows_affected = result.rowcount
    print(f"Updated {rows_affected} rows")

Delete Query:

with db.transaction() as conn:
    result = conn.execute(
        "DELETE FROM sessions WHERE expires_at < :now",
        {"now": datetime.utcnow()}
    )
    
    deleted_count = result.rowcount
    print(f"Deleted {deleted_count} expired sessions")

4.3 ORM Support¶

from sqlalchemy.orm import Session, DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(255))

# Get ORM session
with db.get_session() as session:
    # Query with ORM
    users = session.query(User).filter(User.username.like("john%")).all()
    
    # Create new record
    new_user = User(username="jane_doe", email="jane@example.com")
    session.add(new_user)
    session.commit()
    
    # Update
    user = session.get(User, 1)
    if user:
        user.email = "newemail@example.com"
        session.commit()

5. Backend PostgreSQL¶

5.1 Initialization¶

from structum_lab.plugins.database import PostgresDatabase

# From config
db = PostgresDatabase.from_config()

# Manual
db = PostgresDatabase(
    url="postgresql://user:pass@localhost/mydb",
    pool_size=10,
    min_size=2,  # Minimum connections to maintain
    max_size=20,  # Maximum connections
    timeout=5.0
)

5.2 Query Execution¶

with db.transaction() as conn:
    # Execute with parameters
    result = conn.execute(
        "SELECT * FROM users WHERE id = %s",
        (user_id,)  # Note: tuple for parameters
    )
    
    # Fetch results
    row = result.fetchone()
    if row:
        print(f"User: {row['username']}")

5.3 Async Support¶

import asyncio
from structum_lab.plugins.database import AsyncPostgresDatabase

async def main():
    db = AsyncPostgresDatabase.from_config()
    
    async with db.transaction() as conn:
        result = await conn.execute(
            "SELECT * FROM users WHERE id = $1",
            [user_id]
        )
        
        user = await result.fetchone()
        print(f"User: {user['username']}")
    
    await db.shutdown()

asyncio.run(main())

6. Connection Pooling¶

6.1 Pool Configuration¶

db = SQLAlchemyDatabase(
    url="postgresql://...",
    pool_size=10,          # Normal connections
    max_overflow=20,       # Extra connections under load
    pool_timeout=30,       # Wait time for connection (seconds)
    pool_recycle=3600,     # Recycle after 1 hour
    pool_pre_ping=True     # Test before use
)

6.2 Pool Monitoring¶

# Get pool status
info = db.get_pool_info()

print(f"Checked out: {info['checked_out']}")
print(f"Available: {info['available']}")
print(f"Total: {info['total']}")
print(f"Overflow: {info['overflow']}")

# Log pool stats
from structum_lab.plugins.observability import get_logger

logger = get_logger(__name__)
logger.info("Database pool status", **info)

6.3 Pool Exhaustion Handling¶

from sqlalchemy.exc import TimeoutError

try:
    with db.transaction() as conn:
        # Query execution
        pass
except TimeoutError:
    logger.error("Database pool exhausted - increase pool_size or check for connection leaks")
    # Alert monitoring
    metrics.increment("database.pool_exhaustion")

7. Transactions¶

7.1 Basic Transaction¶

with db.transaction() as conn:
    # All queries in this block are part of single transaction
    conn.execute("INSERT INTO users (...) VALUES (...)")
    conn.execute("INSERT INTO profiles (...) VALUES (...)")
    
    # Automatic COMMIT on success
    # Automatic ROLLBACK on exception

7.2 Explicit Commit/Rollback¶

conn = db.get_connection()
try:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    
    conn.commit()
except Exception as e:
    conn.rollback()
    logger.error(f"Transaction failed: {e}")
    raise
finally:
    conn.close()

7.3 Savepoints¶

with db.transaction() as conn:
    # Main transaction
    conn.execute("INSERT INTO orders (...) VALUES (...)")
    
    # Savepoint for partial rollback
    savepoint = conn.execute("SAVEPOINT sp1")
    
    try:
        conn.execute("INSERT INTO order_items (...) VALUES (...)")
    except Exception:
        # Rollback only to savepoint
        conn.execute("ROLLBACK TO SAVEPOINT sp1")
        logger.warning("Order items insert failed, continuing with order")
    
    # Main transaction still commits

7.4 Nested Transactions¶

def create_user_with_profile(username: str, email: str):
    """Transactional function."""
    with db.transaction() as conn:
        # Insert user
        result = conn.execute(
            "INSERT INTO users (username, email) VALUES (:u, :e) RETURNING id",
            {"u": username, "e": email}
        )
        user_id = result.fetchone()["id"]
        
        # Insert profile (nested transaction)
        create_profile(conn, user_id)
        
        return user_id

def create_profile(conn, user_id: int):
    """Nested transactional function."""
    conn.execute(
        "INSERT INTO profiles (user_id, bio) VALUES (:id, :bio)",
        {"id": user_id, "bio": "New user"}
    )

8. Health Checks¶

8.1 Basic Health Check¶

from structum_lab.plugins.database import HealthStatus

# Perform health check
result = db.health_check()

print(f"Status: {result.status}")           # HealthStatus.HEALTHY
print(f"Message: {result.message}")         # "Database connection OK"
print(f"Latency: {result.latency_ms}ms")   # 2.5

8.2 Health Check Endpoint¶

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health():
    """Health check endpoint."""
    db_health = db.health_check()
    
    return {
        "status": "healthy" if db_health.status == HealthStatus.HEALTHY else "unhealthy",
        "database": {
            "status": db_health.status.value,
            "latency_ms": db_health.latency_ms,
            "message": db_health.message
        }
    }

8.3 Detailed Health Monitoring¶

def monitor_database_health():
    """Periodic health monitoring."""
    result = db.health_check()
    
    # Log metrics
    metrics.gauge("database.latency_ms", result.latency_ms)
    
    if result.status != HealthStatus.HEALTHY:
        logger.error(
            "Database unhealthy",
            status=result.status.value,
            message=result.message,
            latency_ms=result.latency_ms
        )
        
        # Alert
        alert_service.send_alert(
            "Database Health Check Failed",
            f"Status: {result.status.value}\nLatency: {result.latency_ms}ms"
        )
    else:
        logger.debug(f"Database healthy - latency: {result.latency_ms}ms")

9. Configuration¶

9.1 Complete Configuration Example¶