Marketplace sqlmodel-crud

install

source · Clone the upstream repo

git clone https://github.com/aiskillstore/marketplace

Claude Code · Install into ~/.claude/skills/

T=$(mktemp -d) && git clone --depth=1 https://github.com/aiskillstore/marketplace "$T" && mkdir -p ~/.claude/skills && cp -r "$T/skills/awais68/sqlmodel-crud" ~/.claude/skills/aiskillstore-marketplace-sqlmodel-crud && rm -rf "$T"

manifest: skills/awais68/sqlmodel-crud/SKILL.md

SQLModel CRUD Skill

Overview

Expert guidance for SQLModel database models and CRUD operations, including Pydantic integration, async session management, query building with joins, and relationship configuration for ERP entities like Student, Fee, and Attendance.

When This Skill Applies

This skill triggers when users request:

Models: "Student model", "SQLModel", "database model", "table=True"
CRUD Operations: "Create student", "Read fees", "Update attendance", "Delete record"
Query Building: "Query with join", "select statement", "where clause", "pagination"
Relationships: "ForeignKey", "back_populates", "relationship", "linked models"
Validation: "Pydantic validators", "field constraints", "indexes"

Core Rules

1. Models: table=True with Pydantic Validation

# models/student.py
from sqlmodel import SQLModel, Field, Relationship
from datetime import datetime
from typing import Optional, List
from enum import Enum

class StudentRole(str, Enum):
    STUDENT = "student"
    TEACHER = "teacher"
    ADMIN = "admin"

class Student(SQLModel, table=True):
    """Student model with relationships to fees and attendance"""
    id: Optional[str] = Field(default=None, primary_key=True)
    name: str = Field(min_length=2, max_length=100, index=True)
    email: str = Field(unique=True, index=True)
    phone: Optional[str] = Field(default=None, pattern=r'^\+?[\d\s-]+$')
    password_hash: str
    role: StudentRole = Field(default=StudentRole.STUDENT)
    class_id: Optional[str] = Field(default=None, foreign_key="class.id", index=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    # Relationships
    class_obj: Optional["Class"] = Relationship(back_populates="students")
    fees: List["Fee"] = Relationship(back_populates="student")
    attendance: List["Attendance"] = Relationship(back_populates="student")

    class Config:
        from_attributes = True

class Class(SQLModel, table=True):
    """Class model for organizing students"""
    id: Optional[str] = Field(default=None, primary_key=True)
    name: str = Field(min_length=2, max_length=50)
    grade_level: int = Field(ge=1, le=12)
    academic_year: str = Field(max_length=9, pattern=r'^\d{4}-\d{4}$')
    created_at: datetime = Field(default_factory=datetime.utcnow)

    # Relationships
    students: List[Student] = Relationship(back_populates="class_obj")

class Fee(SQLModel, table=True):
    """Fee model linked to students"""
    id: Optional[str] = Field(default=None, primary_key=True)
    student_id: str = Field(foreign_key="student.id", index=True)
    amount: float = Field(gt=0)
    description: str = Field(max_length=500)
    status: str = Field(default="pending", pattern=r'^(pending|paid|overdue|waived)$')
    due_date: datetime
    paid_date: Optional[datetime] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    # Relationships
    student: Optional[Student] = Relationship(back_populates="fees")

class Attendance(SQLModel, table=True):
    """Attendance model linked to students"""
    id: Optional[str] = Field(default=None, primary_key=True)
    student_id: str = Field(foreign_key="student.id", index=True)
    date: datetime = Field(index=True)
    status: str = Field(pattern=r'^(present|absent|late|excused)$')
    notes: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow)

    # Relationships
    student: Optional[Student] = Relationship(back_populates="attendance")

Requirements:

Use
```
table=True
```
for database tables
Add
```
index=True
```
for frequently queried fields
Use
```
unique=True
```
for email and other unique fields
Add
```
foreign_key
```
for relationships
Use Pydantic validators (min_length, max_length, pattern, ge, gt)
Use
```
Relationship
```
with
```
back_populates
```
for bidirectional links
Include
```
created_at
```
and
```
updated_at
```
timestamps

2. CRUD Operations: Async Session Management

# crud/student.py
from sqlmodel import select, and_
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Optional
from datetime import datetime

from models.student import Student
from schemas.student import StudentCreate, StudentUpdate

class StudentCRUD:
    """CRUD operations for Student model"""

    @staticmethod
    async def create(db: AsyncSession, student_data: StudentCreate) -> Student:
        """Create a new student"""
        student = Student(
            name=student_data.name,
            email=student_data.email,
            phone=student_data.phone,
            password_hash=student_data.password_hash,
            role=student_data.role,
            class_id=student_data.class_id,
        )
        db.add(student)
        await db.commit()
        await db.refresh(student)
        return student

    @staticmethod
    async def get_by_id(db: AsyncSession, student_id: str) -> Optional[Student]:
        """Get student by ID"""
        result = await db.execute(
            select(Student).where(Student.id == student_id)
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def get_by_email(db: AsyncSession, email: str) -> Optional[Student]:
        """Get student by email"""
        result = await db.execute(
            select(Student).where(Student.email == email)
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def get_all(
        db: AsyncSession,
        skip: int = 0,
        limit: int = 100,
        class_id: Optional[str] = None,
        role: Optional[str] = None,
    ) -> List[Student]:
        """Get all students with pagination and filters"""
        query = select(Student)

        if class_id:
            query = query.where(Student.class_id == class_id)
        if role:
            query = query.where(Student.role == role)

        query = query.offset(skip).limit(limit).order_by(Student.created_at.desc())

        result = await db.execute(query)
        return result.scalars().all()

    @staticmethod
    async def update(
        db: AsyncSession,
        student_id: str,
        student_data: StudentUpdate,
    ) -> Optional[Student]:
        """Update a student"""
        student = await StudentCRUD.get_by_id(db, student_id)
        if not student:
            return None

        update_data = student_data.model_dump(exclude_unset=True)
        for field, value in update_data.items():
            setattr(student, field, value)

        student.updated_at = datetime.utcnow()
        await db.commit()
        await db.refresh(student)
        return student

    @staticmethod
    async def delete(db: AsyncSession, student_id: str) -> bool:
        """Delete a student"""
        student = await StudentCRUD.get_by_id(db, student_id)
        if not student:
            return False

        await db.delete(student)
        await db.commit()
        return True

    @staticmethod
    async def count(
        db: AsyncSession,
        class_id: Optional[str] = None,
        role: Optional[str] = None,
    ) -> int:
        """Count students with filters"""
        query = select(Student)

        if class_id:
            query = query.where(Student.class_id == class_id)
        if role:
            query = query.where(Student.role == role)

        result = await db.execute(query)
        return len(result.scalars().all())

Requirements:

Use
```
AsyncSession
```
for async database operations

Use

select()

where()

offset()

limit()

order_by()

Return
```
Optional[T]
```
for single items,
```
List[T]
```
for lists
Use
```
scalar_one_or_none()
```
for single results
Use
```
scalars().all()
```
for list results
Handle transactions with commit/rollback
Update
```
updated_at
```
timestamp on modifications

3. Query Building: Joins, Filters, Pagination

# queries/advanced.py
from sqlmodel import select, and_, or_, func
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional

from models.student import Student, Fee, Attendance, Class

class StudentQueries:
    """Advanced student queries with joins"""

    @staticmethod
    async def get_student_with_fees(
        db: AsyncSession,
        student_id: str,
    ) -> Optional[Student]:
        """Get student with all their fees"""
        result = await db.execute(
            select(Student)
            .where(Student.id == student_id)
            .options(
                # Eager load fees relationship
                selectinload(Student.fees)
            )
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def get_students_with_pending_fees(
        db: AsyncSession,
        skip: int = 0,
        limit: int = 100,
    ) -> List[Student]:
        """Get students with pending fee balance"""
        # Subquery to find students with pending fees
        subquery = (
            select(Fee.student_id)
            .where(Fee.status == "pending")
            .distinct()
        )

        result = await db.execute(
            select(Student)
            .where(Student.id.in_(subquery))
            .offset(skip)
            .limit(limit)
            .order_by(Student.name)
        )
        return result.scalars().all()

    @staticmethod
    async def get_student_attendance_summary(
        db: AsyncSession,
        student_id: str,
        month: int,
        year: int,
    ) -> dict:
        """Get attendance summary for a student in a month"""
        from datetime import datetime

        start_date = datetime(year, month, 1)
        end_date = datetime(year, month + 1, 1) if month < 12 else datetime(year + 1, 1, 1)

        result = await db.execute(
            select(
                func.count().label("total_days"),
                func.sum(
                    case((Attendance.status == "present", 1), else_=0)
                ).label("present_days"),
                func.sum(
                    case((Attendance.status == "absent", 1), else_=0)
                ).label("absent_days"),
                func.sum(
                    case((Attendance.status == "late", 1), else_=0)
                ).label("late_days"),
            )
            .where(
                and_(
                    Attendance.student_id == student_id,
                    Attendance.date >= start_date,
                    Attendance.date < end_date,
                )
            )
        )
        row = result.one()
        return {
            "total_days": row.total_days or 0,
            "present_days": row.present_days or 0,
            "absent_days": row.absent_days or 0,
            "late_days": row.late_days or 0,
        }

    @staticmethod
    async def search_students(
        db: AsyncSession,
        search_term: str,
        skip: int = 0,
        limit: int = 100,
    ) -> List[Student]:
        """Search students by name or email"""
        search_pattern = f"%{search_term}%"

        result = await db.execute(
            select(Student)
            .where(
                or_(
                    Student.name.ilike(search_pattern),
                    Student.email.ilike(search_pattern),
                )
            )
            .offset(skip)
            .limit(limit)
            .order_by(Student.name)
        )
        return result.scalars().all()

    @staticmethod
    async def get_students_by_class(
        db: AsyncSession,
        class_id: str,
        include_inactive: bool = False,
    ) -> List[Student]:
        """Get all students in a class"""
        query = select(Student).where(Student.class_id == class_id)

        if not include_inactive:
            # Assuming there's an is_active field
            query = query.where(Student.is_active == True)

        query = query.order_by(Student.name)

        result = await db.execute(query)
        return result.scalars().all()

    @staticmethod
    async def get_fee_statistics(
        db: AsyncSession,
        class_id: Optional[str] = None,
    ) -> dict:
        """Get fee statistics, optionally filtered by class"""
        base_query = select(Fee)

        if class_id:
            # Join with Student to filter by class
            base_query = (
                select(Fee)
                .join(Student, Student.id == Fee.student_id)
                .where(Student.class_id == class_id)
            )

        result = await db.execute(
            select(
                func.count().label("total_fees"),
                func.sum(Fee.amount).label("total_amount"),
                func.sum(
                    case((Fee.status == "paid", Fee.amount), else_=0)
                ).label("total_collected"),
                func.sum(
                    case((Fee.status == "pending", Fee.amount), else_=0)
                ).label("total_pending"),
            )
        )
        row = result.one()
        return {
            "total_fees": row.total_fees or 0,
            "total_amount": float(row.total_amount or 0),
            "total_collected": float(row.total_collected or 0),
            "total_pending": float(row.total_pending or 0),
        }

Requirements:

Use
```
selectinload()
```
for eager loading relationships
Use
```
subquery
```
for complex filtering
Use
```
and_()
```
/
```
or_()
```
for compound conditions
Use
```
func.count()
```
,
```
func.sum()
```
for aggregations
Use
```
case()
```
for conditional aggregation
Use
```
ilike()
```
for case-insensitive search
Use
```
order_by()
```
for sorting
Use
```
offset()
```
/
```
limit()
```
for pagination

4. Relationships: ForeignKey and back_populates

# models/relationships.py
from sqlmodel import SQLModel, Field, Relationship
from datetime import datetime
from typing import Optional, List

class Parent(SQLModel, table=True):
    """Parent model linked to students"""
    id: Optional[str] = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)
    phone: Optional[str]

    # One-to-Many: Parent can have multiple students
    students: List["Student"] = Relationship(
        back_populates="parent",
        link_model="student_parent_link"  # Many-to-many through table
    )

class StudentParentLink(SQLModel, table=True):
    """Many-to-many link table for Student and Parent"""
    student_id: str = Field(foreign_key="student.id", primary_key=True)
    parent_id: str = Field(foreign_key="parent.id", primary_key=True)
    relationship_type: str = Field(default="father")  # father, mother, guardian

# Updated Student with many-to-many relationship
class Student(SQLModel, table=True):
    id: Optional[str] = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)

    # Many-to-Many: Students can have multiple parents
    parents: List[Parent] = Relationship(
        back_populates="students",
        link_model=StudentParentLink
    )

# One-to-One relationship example
class StudentProfile(SQLModel, table=True):
    """One-to-one profile for student"""
    id: Optional[str] = Field(default=None, primary_key=True)
    student_id: str = Field(foreign_key="student.id", unique=True)
    bio: Optional[str] = None
    avatar_url: Optional[str]
    preferences: Optional[str] = None  # JSON string for flexible data

    student: Optional[Student] = Relationship()

# Self-referential relationship
class Employee(SQLModel, table=True):
    """Employee with manager relationship"""
    id: Optional[str] = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)
    manager_id: Optional[str] = Field(foreign_key="employee.id")

    # Self-referential relationship
    manager: Optional["Employee"] = Relationship(
        back_populates="subordinates",
        sa_relationship_kwargs={"remote_side": "[Employee.id]"}
    )
    subordinates: List["Employee"] = Relationship(back_populates="manager")

Requirements:

Use
```
foreign_key
```
for relational links
Use
```
back_populates
```
for bidirectional relationships
Use
```
unique=True
```
for one-to-one relationships
Use link tables for many-to-many relationships
Use
```
remote_side
```
for self-referential relationships
Use
```
sa_relationship_kwargs
```
for advanced SQLAlchemy options

Output Requirements

Code Files

Models:

```
models/__init__.py
```
```
models/student.py
```
```
models/fee.py
```
```
models/attendance.py
```
```
models/class.py
```

CRUD Operations:

```
crud/__init__.py
```
```
crud/student.py
```
```
crud/fee.py
```
```
crud/attendance.py
```

Advanced Queries:
- ```
queries/__init__.py
```
- ```
queries/advanced.py
```

Integration Requirements

@fastapi-app: Use models in FastAPI routes
@fastapi-app/dependencies: DB session dependency
@api-client: Type-safe responses for frontend

Documentation

PHR: Create Prompt History Record for schema design
ADR: Document relationship choices, indexing strategy
Comments: Document complex queries and relationships

Workflow

Design Schema
- Identify entities (Student, Fee, Attendance, etc.)
- Define relationships (one-to-many, many-to-many)
- Add constraints (unique, foreign keys, indexes)
Create Models
- Define SQLModel classes with
```
table=True
```
- Add fields with types and validators
- Configure relationships with
```
back_populates
```
Build CRUD Operations
- Implement create, read, update, delete
- Add pagination and filtering
- Handle async sessions
Create Advanced Queries
- Implement joins and subqueries
- Add aggregations and summaries
- Search and filter functionality
Test and Optimize
- Test all CRUD operations
- Verify relationships work correctly
- Optimize slow queries with indexes

Quality Checklist

Before completing any SQLModel implementation:

Common Patterns

Student Model with CRUD

# models/student.py
class Student(SQLModel, table=True):
    id: Optional[str] = Field(default=None, primary_key=True)
    name: str = Field(min_length=2, max_length=100, index=True)
    email: str = Field(unique=True, index=True)
    is_active: bool = Field(default=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    fees: List["Fee"] = Relationship(back_populates="student")

CRUD Fees

# crud/fee.py
class FeeCRUD:
    @staticmethod
    async def create(db: AsyncSession, fee_data: FeeCreate) -> Fee:
        fee = Fee(**fee_data.model_dump())
        db.add(fee)
        await db.commit()
        await db.refresh(fee)
        return fee

    @staticmethod
    async def get_by_id(db: AsyncSession, fee_id: str) -> Optional[Fee]:
        result = await db.execute(
            select(Fee).where(Fee.id == fee_id)
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def get_pending_by_student(db: AsyncSession, student_id: str) -> List[Fee]:
        result = await db.execute(
            select(Fee)
            .where(
                and_(
                    Fee.student_id == student_id,
                    Fee.status == "pending"
                )
            )
            .order_by(Fee.due_date)
        )
        return result.scalars().all()

Query with Join

# Get students with their class name using join
async def get_students_with_class(db: AsyncSession) -> List[dict]:
    result = await db.execute(
        select(
            Student.id,
            Student.name,
            Student.email,
            Class.name.label("class_name"),
        )
        .outerjoin(Class, Student.class_id == Class.id)
        .order_by(Student.name)
    )
    return [dict(row._mapping) for row in result]

Database Setup

# database.py
from sqlmodel import create_engine, SQLModel
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+asyncpg://user:password@localhost:5432/erp_db"
)

# Async engine
async_engine = create_async_engine(DATABASE_URL, echo=True)
async_session_maker = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def init_db():
    """Initialize database tables"""
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

async def get_db() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Migrations with Alembic

# alembic/env.py
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_url
from alembic import context

# Import models for autogenerate
from models.student import Student
from models.fee import Fee
from models.attendance import Attendance

target_metadata = SQLModel.metadata

# alembic revision --autogenerate -m "Initial migration"
# alembic upgrade head

References

SQLModel Documentation: https://sqlmodel.tiangolo.com
SQLAlchemy Relationships: https://docs.sqlalchemy.org/en/20/orm/relationships.html
SQLAlchemy Async: https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
Pydantic Validation: https://docs.pydantic.dev/latest/usage/validators/
Alembic Migrations: https://alembic.sqlalchemy.org/en/latest/