SuperAPI AI Directory

database-schema

Data & analytics

View repository

Updated 10-15 23:42

About this skill

Database Schema Generator

This skill provides comprehensive tools for generating database schemas with proper relations, migrations, and ORM/ODM models for PostgreSQL, MongoDB, and SQLite that integrate seamlessly with FastAPI applications.

When to Use This Skill

Use this skill when you need to:

  • Generate database schemas with proper relationships and constraints
  • Create ORM/ODM models for SQL (SQLAlchemy) or NoSQL (ODMantic/PyMongo) databases
  • Set up database migrations for schema evolution
  • Define proper indexing strategies for performance
  • Generate FastAPI integration patterns for database operations
  • Create database connection pools and session management

Supported Database Types

SQL Databases

  • PostgreSQL: Advanced features, JSON support, full-text search
  • SQLite: Lightweight, file-based, perfect for development/testing
  • MySQL: Traditional SQL with comprehensive feature set (coming soon)

NoSQL Databases

  • MongoDB: Document-based with flexible schema and rich query language

Core Workflow

1. Database Type Selection

  • Choose between PostgreSQL, MongoDB, or SQLite based on requirements
  • Consider factors: scalability, ACID compliance, document flexibility, deployment complexity

2. Schema Design

  • Define entities and their relationships
  • Plan indexes for optimal query performance
  • Consider data normalization vs. denormalization trade-offs

3. Model Generation

  • Create appropriate ORM/ODM models based on database type
  • Define proper field types and constraints
  • Implement relationship mappings

4. Migration Strategy

  • Generate migration files for schema evolution
  • Plan rollback strategies for safe deployments
  • Consider data migration needs

5. FastAPI Integration

  • Set up database connection pools
  • Implement dependency injection for database sessions
  • Create proper error handling for database operations

Database-Specific Patterns

PostgreSQL Schema Generation

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Index
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from database.base import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    # Relationships
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")


# PostgreSQL-specific indexes
Index("idx_user_email", "email", unique=True)
Index("idx_user_created_at", "created_at")

MongoDB Schema Generation (ODMantic)

from odmantic import Model, Field, Index
from datetime import datetime
from typing import List, Optional


class User(Model):
    email: str = Field(unique=True, regex=r"^[\w\.-]+@[\w\.-]+\.\w+$")
    name: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        collection = "users"
        indexes = [Index("email", unique=True), Index("created_at")]


class Post(Model):
    title: str
    content: str
    author_id: str = Field(foreign_key="User.id")
    created_at: datetime = Field(default_factory=datetime.utcnow)

    # Embedded relationships in MongoDB
    tags: List[str] = []
    metadata: Optional[dict] = {}

    class Config:
        collection = "posts"
        indexes = [Index("author_id"), Index("created_at"), Index("tags")]

SQLite Schema Generation

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from database.base import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

    posts = relationship("Post", back_populates="author")

Migration Patterns

Alembic Migration Example

"""Add user profile fields

Revision ID: abc123def456
Revises: 7d5c8b1a2c3d
Create Date: 2023-10-15 10:30:00.000000

"""

from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = "abc123def456"
down_revision = "7d5c8b1a2c3d"
branch_labels = None
depends_on = None


def upgrade():
    # Add new columns
    op.add_column("users", sa.Column("bio", sa.Text(), nullable=True))
    op.add_column("users", sa.Column("avatar_url", sa.String(500), nullable=True))
    op.add_column(
        "users", sa.Column("is_verified", sa.Boolean(), nullable=True, default=False)
    )

    # Create indexes
    op.create_index("ix_users_bio", "users", ["bio"])
    op.create_index("ix_users_is_verified", "users", ["is_verified"])


def downgrade():
    # Remove columns (in reverse order)
    op.drop_index("ix_users_is_verified")
    op.drop_index("ix_users_bio")
    op.drop_column("users", "is_verified")
    op.drop_column("users", "avatar_url")
    op.drop_column("users", "bio")

FastAPI Integration Patterns

Database Dependency

from fastapi import Depends
from sqlalchemy.orm import Session
from database.session import get_db


async def get_current_user(
    token: str = Security(oauth2_scheme), db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = db.query(User).filter(User.email == email).first()
    if user is None:
        raise credentials_exception
    return user

Best Practices

Performance Optimization

  • Use proper indexing strategies
  • Implement connection pooling
  • Use eager loading for related data when needed
  • Consider caching strategies for read-heavy operations

Security Considerations

  • Sanitize all database inputs
  • Use parameterized queries to prevent injection
  • Implement proper authentication and authorization
  • Encrypt sensitive data at rest

Scalability Patterns

  • Plan for database sharding if needed
  • Use read replicas for read-heavy operations
  • Implement proper database connection management
  • Consider database-specific optimization techniques

Advanced Features

Relationship Handling

  • One-to-Many relationships with proper cascading
  • Many-to-Many relationships with join tables
  • One-to-One relationships for specialized use cases
  • Self-referencing relationships for hierarchical data

Data Validation

  • Database-level constraints
  • Application-level validation through ORM/ODM
  • Custom validation functions
  • Data integrity checks

References