Modulo Database (structum_lab.database)¶

from typing import Protocol

class DatabaseInterface(Protocol):
    def transaction(self) -> TransactionContext:
        ...

# NO inheritance required
class MyDatabase:  # Automatically compatible!
    def transaction(self) -> TransactionContext:
        # Implementation
        return MyTransactionContext()

# Type checker verifies compatibility
db: DatabaseInterface = MyDatabase()  # ✓ Works!

from structum_lab.database import DatabaseInterface

def fetch_user(db: DatabaseInterface, user_id: int) -> dict:
    """Type-safe database function."""
    with db.transaction() as conn:
        result = conn.execute(
            "SELECT * FROM users WHERE id = :id",
            {"id": user_id}
        )
        return result.fetchone()

# mypy verifies db has transaction() at compile-time

from typing import Protocol, ContextManager

class DatabaseInterface(Protocol):
    """
    Protocol for database connectivity.
    
    Implementations must provide:
    - Connection management
    - Transaction support
    - Health monitoring
    """
    
    def transaction(self) -> ContextManager[ConnectionInterface]:
        """
        Get transaction context manager.
        
        Usage:
            with db.transaction() as conn:
                conn.execute("SELECT ...")
        
        Returns:
            Context manager that:
            - Acquires connection from pool
            - Commits on success
            - Rollbacks on exception
            - Returns connection to pool
        """
        ...
    
    def get_connection(self) -> ConnectionInterface:
        """
        Get raw connection (manual management).
        
        Warning:
            MUST call connection.close() manually.
            Prefer transaction() context manager.
        """
        ...
    
    def health_check(self) -> HealthResult:
        """
        Perform database health check.
        
        Returns:
            HealthResult with status, message, latency
        """
        ...
    
    def shutdown(self) -> None:
        """
        Gracefully shutdown database connections.
        
        Call at application shutdown to:
        - Close all pool connections
        - Release resources
        """
        ...

from structum_lab.database import get_database

# Get database instance
db = get_database()

# Execute query in transaction
with db.transaction() as conn:
    # Insert
    result = conn.execute(
        "INSERT INTO users (username, email) VALUES (:u, :e) RETURNING id",
        {"u": "john_doe", "e": "john@example.com"}
    )
    new_id = result.fetchone()["id"]
    
    # Query
    result = conn.execute(
        "SELECT * FROM users WHERE id = :id",
        {"id": new_id}
    )
    user = result.fetchone()
    
    # Auto-commit on success
    # Auto-rollback on exception

# Check health
health = db.health_check()
if health.status == HealthStatus.HEALTHY:
    print(f"Database OK - latency: {health.latency_ms}ms")

from typing import Protocol, Any, Optional

class ConnectionInterface(Protocol):
    """
    Protocol for active database connection.
    
    Represents a single connection that can execute queries.
    """
    
    def execute(self, query: str, params: dict = None) -> ResultProxy:
        """
        Execute SQL query.
        
        Args:
            query: SQL statement (use :param for placeholders)
            params: Dictionary of parameter values
        
        Returns:
            ResultProxy for fetching results
        
        Example:
            result = conn.execute(
                "SELECT * FROM users WHERE id = :id",
                {"id": 123}
            )
            user = result.fetchone()
        """
        ...
    
    def commit(self) -> None:
        """Commit current transaction."""
        ...
    
    def rollback(self) -> None:
        """Rollback current transaction."""
        ...
    
    def close(self) -> None:
        """Close connection and return to pool."""
        ...

class ResultProxy(Protocol):
    """Protocol for query results."""
    
    def fetchone(self) -> Optional[dict]:
        """
        Fetch single row.
        
        Returns:
            Dictionary with column names as keys
            None if no rows
        """
        ...
    
    def fetchall(self) -> list[dict]:
        """
        Fetch all remaining rows.
        
        Returns:
            List of dictionaries
        """
        ...
    
    def fetchmany(self, size: int = None) -> list[dict]:
        """
        Fetch batch of rows.
        
        Args:
            size: Number of rows (default: arraysize)
        
        Returns:
            List of dictionaries
        """
        ...
    
    @property
    def rowcount(self) -> int:
        """Number of rows affected by query."""
        ...

with db.transaction() as conn:
    result = conn.execute("SELECT * FROM users WHERE active = :active", {"active": True})
    
    # Single row
    user = result.fetchone()
    if user:
        print(f"User: {user['username']}")
    
    # All rows
    users = result.fetchall()
    for user in users:
        print(user['username'])
    
    # Batch fetch
    batch = result.fetchmany(size=100)

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

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

from typing import Protocol, ContextManager

class TransactionContext(Protocol):
    """Context manager for database transactions."""
    
    def __enter__(self) -> ConnectionInterface:
        """
        Acquire connection and start transaction.
        
        Returns:
            Connection for executing queries
        """
        ...
    
    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        """
        Complete transaction and release connection.
        
        Behavior:
            - If exc_type is None: COMMIT
            - If exception occurred: ROLLBACK
            - Always: Return connection to pool
        """
        ...

# Success case
with db.transaction() as conn:
    conn.execute("INSERT INTO orders (...) VALUES (...)")
    conn.execute("INSERT INTO order_items (...) VALUES (...)")
    # Automatic COMMIT on clean exit

# Failure case
try:
    with db.transaction() as conn:
        conn.execute("INSERT INTO orders (...) VALUES (...)")
        raise ValueError("Something went wrong!")
        conn.execute("INSERT INTO order_items (...) VALUES (...)")
except ValueError:
    # Automatic ROLLBACK - no partial data
    pass

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 (same transaction)
        conn.execute(
            "INSERT INTO profiles (user_id, bio) VALUES (:id, :bio)",
            {"id": user_id, "bio": "New user"}
        )
        
        # Both inserts commit atomically
        return user_id

from enum import Enum
from dataclasses import dataclass

class HealthStatus(Enum):
    """Health check status enumeration."""
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"

@dataclass
class HealthResult:
    """Database health check result."""
    
    status: HealthStatus      # Overall health status
    message: str              # Human-readable message
    latency_ms: float         # Query latency in milliseconds
    details: dict = None      # Optional detailed info

from structum_lab.database import get_database, HealthStatus

db = get_database()
health = db.health_check()

# Check status
if health.status == HealthStatus.HEALTHY:
    print(f"✓ Database healthy - latency: {health.latency_ms}ms")
elif health.status == HealthStatus.DEGRADED:
    print(f"⚠ Database degraded: {health.message}")
else:
    print(f"✗ Database unhealthy: {health.message}")

# Log metrics
from structum_lab.logging import get_logger
logger = get_logger(__name__)

logger.info(
    "Database health check",
    status=health.status.value,
    latency_ms=health.latency_ms,
    message=health.message
)

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health_check():
    """Application health endpoint."""
    db = get_database()
    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
        }
    }

from structum_lab.database import set_database
from structum_lab.plugins.database import SQLAlchemyDatabase

# Initialize and register
db = SQLAlchemyDatabase.from_config()
set_database(db)

# Now available globally
from structum_lab.database import get_database
db = get_database()

from structum_lab.plugins.database import SQLAlchemyDatabase

# Method 1: Direct configuration
db = SQLAlchemyDatabase(
    url="postgresql://user:pass@localhost/mydb",
    pool_size=10
)

# Method 2: From config provider (recommended)
db = SQLAlchemyDatabase.from_config()
# Reads: config.get("database.url"), config.get("database.pool_size"), etc.

# Method 3: Custom config namespace
primary_db = SQLAlchemyDatabase.from_config("primary_db")
replica_db = SQLAlchemyDatabase.from_config("replica_db")

from structum_lab.database import get_database, set_database
from structum_lab.plugins.database import SQLAlchemyDatabase

# Setup (once at application startup)
def setup_database():
    db = SQLAlchemyDatabase.from_config()
    set_database(db)

# Usage (anywhere in application)
def get_user(user_id: int):
    db = get_database()
    
    with db.transaction() as conn:
        result = conn.execute(
            "SELECT * FROM users WHERE id = :id",
            {"id": user_id}
        )
        return result.fetchone()

from structum_lab.database import DatabaseInterface

class UserRepository:
    """Repository with injected database dependency."""
    
    def __init__(self, db: DatabaseInterface):
        self.db = db
    
    def find_by_id(self, user_id: int) -> dict | None:
        with self.db.transaction() as conn:
            result = conn.execute(
                "SELECT * FROM users WHERE id = :id",
                {"id": user_id}
            )
            return result.fetchone()

# With DI container
from structum_lab.plugins.di import StructumContainer

class AppContainer(StructumContainer):
    database = providers.Singleton(SQLAlchemyDatabase.from_config)
    
    user_repo = providers.Factory(
        UserRepository,
        db=database
    )

from fastapi import FastAPI, Depends
from structum_lab.database import get_database, DatabaseInterface

app = FastAPI()

# Startup/shutdown
@app.on_event("startup")
async def startup():
    db = SQLAlchemyDatabase.from_config()
    set_database(db)

@app.on_event("shutdown")
async def shutdown():
    db = get_database()
    db.shutdown()

# Dependency
def get_db() -> DatabaseInterface:
    return get_database()

# Route
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: DatabaseInterface = Depends(get_db)):
    with db.transaction() as conn:
        result = conn.execute(
            "SELECT * FROM users WHERE id = :id",
            {"id": user_id}
        )
        user = result.fetchone()
        
        if not user:
            raise HTTPException(status_code=404)
        
        return user

def get_database() -> DatabaseInterface:
    """
    Get global database singleton.
    
    Raises:
        RuntimeError if set_database() not called
    """
    ...

def set_database(db: DatabaseInterface) -> None:
    """
    Set global database instance (call once at startup).
    
    Args:
        db: Database implementation instance
    """
    ...

class DatabaseInterface(Protocol):
    def transaction(self) -> ContextManager[ConnectionInterface]:
        ...
    
    def get_connection(self) -> ConnectionInterface:
        ...
    
    def health_check(self) -> HealthResult:
        ...
    
    def shutdown(self) -> None:
        ...

class ConnectionInterface(Protocol):
    def execute(self, query: str, params: dict = None) -> ResultProxy:
        ...
    
    def commit(self) -> None:
        ...
    
    def rollback(self) -> None:
        ...
    
    def close(self) -> None:
        ...

class ResultProxy(Protocol):
    def fetchone(self) -> Optional[dict]:
        ...
    
    def fetchall(self) -> list[dict]:
        ...
    
    def fetchmany(self, size: int = None) -> list[dict]:
        ...
    
    @property
    def rowcount(self) -> int:
        ...

class HealthStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"

@dataclass
class HealthResult:
    status: HealthStatus
    message: str
    latency_ms: float
    details: dict = None