Database Operations & Management

This skill provides comprehensive database patterns and best practices for 2025, focusing on async operations, performance optimization, and production-ready configurations that work across different database systems.

When to Use This Skill

Use this skill when you need to:

• Design database schemas with SQLModel/SQLAlchemy
• Implement async database operations
• Optimize database performance
• Set up connection pooling
• Create and manage database migrations
• Handle complex relationships and queries
• Set up production database configurations
• Monitor and troubleshoot database issues

Core Database Patterns

1. Async Connection Management

# Database connection setup with connection pooling
import asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool
from contextlib import asynccontextmanager
import os

# Environment-based configuration
DATABASE_CONFIG = {
    "development": {
        "url": "sqlite+aiosqlite:///./app.db",
        "poolclass": NullPool,
        "echo": True
    },
    "production": {
        "url": os.getenv("DATABASE_URL", "postgresql+asyncpg://user:pass@localhost/db"),
        "pool_size": 30,
        "max_overflow": 40,
        "pool_pre_ping": True,
        "pool_recycle": 3600,
        "echo": False
    }
}

class DatabaseManager:
    """Universal database manager for async operations"""

    def __init__(self, env: str = "development"):
        config = DATABASE_CONFIG[env]
        self.engine = create_async_engine(**config)
        self.async_session = async_sessionmaker(
            self.engine,
            class_=AsyncSession,
            expire_on_commit=False
        )

    @asynccontextmanager
    async def get_session(self):
        """Get async database session with proper cleanup"""
        async with self.async_session() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise
            finally:
                await session.close()

    async def create_tables(self, metadata):
        """Create all tables from metadata"""
        async with self.engine.begin() as conn:
            await conn.run_sync(metadata.create_all)

    async def close(self):
        """Close database connections"""
        await self.engine.dispose()

# Singleton instance
db_manager = DatabaseManager(os.getenv("ENV", "development"))

2. Base Model with Best Practices

# models/base.py
from sqlmodel import SQLModel, Field, DateTime, func
from typing import Optional
from datetime import datetime
from sqlalchemy import Column, DateTime as SQLDateTime, Index

class TimestampMixin(SQLModel):
    """Mixin for timestamp fields"""

    created_at: datetime = Field(
        default_factory=datetime.utcnow,
        sa_column=Column(
            SQLDateTime(timezone=True),
            server_default=func.now(),
            nullable=False
        )
    )

    updated_at: Optional[datetime] = Field(
        default=None,
        sa_column=Column(
            SQLDateTime(timezone=True),
            onupdate=func.now(),
            nullable=True
        )
    )

class SoftDeleteMixin(SQLModel):
    """Mixin for soft delete functionality"""

    deleted_at: Optional[datetime] = None
    is_deleted: bool = Field(default=False)

class BaseModel(SQLModel, TimestampMixin):
    """Base model with common fields and patterns"""

    id: Optional[int] = Field(default=None, primary_key=True)

    class Config:
        # Enable Pydantic's strict mode
        strict = True
        # Validate defaults
        validate_assignment = True
        # Use enum values
        use_enum_values = True

# Indexes for common queries
Index("idx_base_created_at", BaseModel.created_at)

3. Advanced Model Patterns

# models/examples.py
from enum import Enum
from typing import Optional, List
from sqlalchemy import Column, String, Text, Boolean, Integer, Index, ForeignKey, UniqueConstraint
from sqlalchemy.orm import relationship
from sqlmodel import SQLModel, Field, Session, select, update, delete

# Enums for type safety
class Status(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    PENDING = "pending"

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class User(BaseModel, table=True):
    """User model with optimized fields and indexes"""

    __tablename__ = "users"

    email: str = Field(
        sa_column=Column(String(255), unique=True, nullable=False, index=True)
    )
    username: str = Field(
        sa_column=Column(String(100), unique=True, nullable=False, index=True)
    )
    full_name: Optional[str] = Field(default=None, max_length=200)
    is_active: bool = Field(default=True, sa_column=Column(Boolean, default=True))

    # Optimized text fields
    bio: Optional[str] = Field(
        default=None,
        sa_column=Column(Text)  # Use Text for longer content
    )

    # Relationships
    tasks: List["Task"] = Relationship(back_populates="owner")

    # Optimized queries
    __table_args__ = (
        Index("idx_user_active_email", "is_active", "email"),
        Index("idx_user_created_at", "created_at"),
    )

class Task(BaseModel, SoftDeleteMixin, table=True):
    """Task model with advanced patterns"""

    __tablename__ = "tasks"

    # Foreign key with index
    owner_id: int = Field(
        foreign_key="users.id",
        sa_column=Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
    )

    # Optimized fields
    title: str = Field(
        sa_column=Column(String(200), nullable=False)
    )
    description: Optional[str] = Field(
        default=None,
        sa_column=Column(Text)
    )

    # Enum fields
    status: Status = Field(default=Status.PENDING)
    priority: Priority = Field(default=Priority.MEDIUM)

    # JSON field for flexible metadata
    metadata: dict = Field(
        default_factory=dict,
        sa_column=Column("metadata", JSON)
    )

    # Relationships
    owner: User = Relationship(back_populates="tasks")
    tags: List["TaskTag"] = Relationship(back_populates="task")

    # Optimized queries
    __table_args__ = (
        Index("idx_task_owner_status", "owner_id", "status"),
        Index("idx_task_priority_created", "priority", "created_at"),
        Index("idx_task_deleted", "is_deleted"),
    )

class Tag(BaseModel, table=True):
    """Tag model for many-to-many relationships"""

    __tablename__ = "tags"

    name: str = Field(
        sa_column=Column(String(50), unique=True, nullable=False, index=True)
    )
    color: Optional[str] = Field(default=None, max_length=7)  # Hex color code

class TaskTag(BaseModel, table=True):
    """Many-to-many relationship table"""

    __tablename__ = "task_tags"

    task_id: int = Field(foreign_key="tasks.id", primary_key=True)
    tag_id: int = Field(foreign_key="tags.id", primary_key=True)

    task: Task = Relationship()
    tag: Tag = Relationship()

4. Repository Pattern for Clean Code

# repositories/base.py
from abc import ABC, abstractmethod
from typing import TypeVar, Generic, List, Optional, Dict, Any
from sqlmodel import SQLModel, Session, select, update, delete, func

ModelType = TypeVar("ModelType", bound=SQLModel)

class BaseRepository(Generic[ModelType], ABC):
    """Base repository with common CRUD operations"""

    def __init__(self, model: type[ModelType]):
        self.model = model

    async def create(self, db: Session, *, obj_in: Dict[str, Any]) -> ModelType:
        """Create a new record"""
        db_obj = self.model(**obj_in)
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    async def get(self, db: Session, id: Any) -> Optional[ModelType]:
        """Get a record by ID"""
        statement = select(sel