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=Truefor database tables - Add
index=Truefor frequently queried fields - Use
unique=Truefor email and other unique fields - Add
foreign_keyfor relationships - Use Pydantic validators (min_length, max_length, pattern, ge, gt)
- Use
Relationshipwithback_populatesfor bidirectional links - Include
created_atandupdated_attimestamps
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
AsyncSessionfor 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_attimestamp on modifications
3. Query Building: Joins, Filters, Pagination
# queries/advanced.py
from sqlmodel import select, and_,