Advanced CRUD in FastAPI with SQL Server

In Basic CRUD in FastAPI you built an API with plain dictionaries and raw SQL. It worked, but it had no validation, returned 200 OK for missing records and repeated the same database code in every endpoint. In this lesson you will build the same Student API the way it is done in real projects.

• SQLAlchemy 2.0 ORM with typed models instead of raw SQL (like Entity Framework Core)
• Pydantic schemas that validate every request and shape every response (like DTOs with data annotations)
• Dependency injection with Depends(get_db): one database session per request, always closed
• Correct HTTP status codes with HTTPException
• Transactions with rollback when a database write fails
• Alembic migrations: create the SQL Server table from Python code (Code First)

Project Structure

Instead of one file, each responsibility gets its own module. This is the standard layout for a FastAPI + SQLAlchemy project:

StudentSQLServerAPI/
│
├── database.py        # engine, session factory, Base class, get_db()
├── models.py          # SQLAlchemy model  → the database table
├── schemas.py         # Pydantic schemas  → request / response JSON
├── main.py            # FastAPI app and CRUD endpoints
│
├── alembic.ini        # Alembic configuration
└── migrations/
    ├── env.py         # tells Alembic where the models and database are
    └── versions/
        └── 68bc5628590a_create_students_table.py

Install the Packages

python -m venv venv
venv\Scripts\activate

pip install fastapi uvicorn sqlalchemy pyodbc alembic

Create an empty database in SQL Server Management Studio. Alembic creates the tables, but not the database itself:

CREATE DATABASE StudentFastAPIDB;

Step 1: database.py, the Connection Layer

from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# 1. SQL Server connection details
connection_string = (
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=DESKTOP-PFD5S9B\\SQLEXPRESS;"
    "DATABASE=StudentFastAPIDB;"
    "Trusted_Connection=yes;"
    "TrustServerCertificate=yes;"
)

# 2. Build the SQLAlchemy connection URL
DATABASE_URL = URL.create(
    "mssql+pyodbc",
    query={"odbc_connect": connection_string}
)

# 3. Create the database engine
engine = create_engine(DATABASE_URL)

# 4. Create a factory for database sessions
SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    autocommit=False
)

# 5. Base class for SQLAlchemy models
class Base(DeclarativeBase):
    pass


# 6. Provide a database session to API endpoints
def get_db():
    db = SessionLocal()

    try:
        yield db
    finally:
        db.close()

Each numbered part has one job:

1. Connection string: a normal ODBC connection string, the same one you would use with pyodbc directly. TrustServerCertificate=yes avoids SSL certificate errors on a local SQL Server.
2. URL.create wraps it in a SQLAlchemy URL. Passing it through odbc_connect means you don't have to URL-encode the backslash, spaces and braces yourself.
3. Engine: owns the connection pool. It is created once when the app starts and reused by every request, so a new physical connection is not opened for each call.
4. SessionLocal: a factory that creates sessions. A session is your unit of work: it tracks the objects you add or change and writes them on commit(), just like DbContext.SaveChanges().
5. Base: every model inherits from it. Base.metadata collects all tables, which Alembic reads later.
6. get_db(): a generator dependency. The code before yield runs before the endpoint, the code in finally runs after the response, even if the endpoint raised an exception. That fixes the connection leak from the basic version.

Never commit a real connection string to Git. In a real project read it from an environment variable, for example os.getenv("DB_CONNECTION"), or from a .env file with pydantic-settings.

Step 2: models.py, the Table

from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column

from database import Base


class Student(Base):
    __tablename__ = "Students"

    Id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True
    )

    Name: Mapped[str] = mapped_column(
        String(100),
        nullable=False
    )

    Age: Mapped[int] = mapped_column(
        Integer,
        nullable=False
    )

    Gender: Mapped[str] = mapped_column(
        String(20),
        nullable=False
    )

This is the modern SQLAlchemy 2.0 style. Mapped[int] gives the attribute a real Python type, so your editor knows student.Age is an int and can autocomplete and type-check it. mapped_column describes the SQL column:

Step 3: schemas.py, Validation with Pydantic

The model describes the table. A schema describes the JSON that goes in and out of the API. Keeping them separate means a client can never set the Id themselves, and you choose exactly which fields are returned.

from pydantic import BaseModel, ConfigDict, Field


# Data received when creating or updating a student
class StudentCreate(BaseModel):
    Name: str = Field(min_length=2, max_length=100)
    Age: int = Field(ge=1, le=120)
    Gender: str = Field(min_length=1, max_length=20)


# Data returned to the client
class StudentResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    Id: int
    Name: str
    Age: int
    Gender: str

• StudentCreate has no Id: the database generates it.
• Field(min_length=2, max_length=100) works like [StringLength(100, MinimumLength = 2)]. The max length matches the String(100) column, so too-long text is rejected before it reaches SQL Server.
• ge=1, le=120 means "greater than or equal to 1" and "less than or equal to 120", like [Range(1, 120)].
• from_attributes=True lets Pydantic read values from a SQLAlchemy object (student.Name) instead of only from a dictionary. Without it, returning a model from an endpoint with response_model fails.

Step 4: Alembic Migrations (Code First)

Instead of writing CREATE TABLE by hand, Alembic compares your models with the database and generates the SQL for you. If you have used EF Core, this is Add-Migration and Update-Database.

Initialize Alembic

Run this once in the project folder. It creates alembic.ini and the migrations folder:

alembic init migrations

Connect Alembic to your models in env.py

Open migrations/env.py and make two changes. First, import your Base, engine and models, and point target_metadata at Base.metadata:

from database import Base, DATABASE_URL, engine
import models

target_metadata = Base.metadata

The line import models looks unused, but it is essential: importing the module is what registers the Student class on Base.metadata. Without it Alembic sees no tables and generates an empty migration.

Second, replace run_migrations_online() so Alembic uses the same engine as the API. Then the connection string lives in one place (database.py) and sqlalchemy.url in alembic.ini can stay empty:

def run_migrations_online():
    with engine.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True
        )

        with context.begin_transaction():
            context.run_migrations()

compare_type=True makes Alembic also detect column type changes, for example String(50) changed to String(100). By default it only detects added and removed columns.

Generate the migration

alembic revision --autogenerate -m "Create Students table"

Alembic creates a file in migrations/versions:

"""Create Students table

Revision ID: 68bc5628590a
Revises:
Create Date: 2026-10-23 15:00:25.887392

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '68bc5628590a'
down_revision: Union[str, Sequence[str], None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """Upgrade schema."""
    op.create_table('Students',
    sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False),
    sa.Column('Name', sa.String(length=100), nullable=False),
    sa.Column('Age', sa.Integer(), nullable=False),
    sa.Column('Gender', sa.String(length=20), nullable=False),
    sa.PrimaryKeyConstraint('Id')
    )


def downgrade() -> None:
    """Downgrade schema."""
    op.drop_table('Students')

upgrade() applies the change and downgrade() undoes it. down_revision = None means this is the first migration; the next one will point back to '68bc5628590a', forming a chain. Always open and read a generated migration before you apply it.

Apply the migration

alembic upgrade head

The Students table now exists in StudentFastAPIDB, together with an alembic_version table where Alembic stores the current revision (like __EFMigrationsHistory).

Step 5: main.py, the CRUD Endpoints

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

from database import get_db
from models import Student
from schemas import StudentCreate, StudentResponse


# Create the FastAPI application
app = FastAPI(title="Student SQL Server CRUD API")

The title is shown at the top of the Swagger page.

1. Create a student: POST /students

@app.post("/students", response_model=StudentResponse)
def create_student(
    student_data: StudentCreate,
    db: Session = Depends(get_db)
):
    new_student = Student(
        Name=student_data.Name,
        Age=student_data.Age,
        Gender=student_data.Gender
    )

    try:
        db.add(new_student)
        db.commit()
        db.refresh(new_student)
        return new_student

    except Exception:
        db.rollback()
        raise HTTPException(
            status_code=500,
            detail="Unable to create student"
        )

There is a lot happening in these few lines:

• student_data: StudentCreate: FastAPI reads the JSON body and validates it against the schema. If it is invalid, your function never runs.
• db: Session = Depends(get_db): FastAPI calls get_db(), injects the session and closes it after the response. This is constructor injection of DbContext in a .NET controller.
• db.add() marks the object for insert; db.commit() runs the INSERT inside a transaction.
• db.refresh(new_student) reloads the row from the database, so new_student.Id now holds the identity value SQL Server generated.
• response_model=StudentResponse converts the SQLAlchemy object into the response JSON and filters out any field not in the schema.
• If anything fails, db.rollback() undoes the transaction and the client gets a clean 500 message instead of a stack trace.

2. Read all students: GET /students

@app.get("/students", response_model=list[StudentResponse])
def get_students(db: Session = Depends(get_db)):
    students = db.query(Student).all()
    return students

db.query(Student).all() runs SELECT ... FROM Students and returns a list of Student objects. list[StudentResponse] tells FastAPI the response is a JSON array, and Swagger documents it that way.

3. Read one student: GET /students/{id}

@app.get("/students/{student_id}", response_model=StudentResponse)
def get_student(student_id: int, db: Session = Depends(get_db)):
    student = db.get(Student, student_id)

    if student is None:
        raise HTTPException(
            status_code=404,
            detail="Student not found"
        )

    return student

db.get(Student, student_id) finds a row by primary key, like DbSet.Find(id). If no row exists, raise HTTPException(status_code=404) stops the function and sends a real 404 Not Found response. That is the fix for the basic version, which answered "not found" with 200 OK.

4. Update a student: PUT /students/{id}

@app.put("/students/{student_id}", response_model=StudentResponse)
def update_student(
    student_id: int,
    student_data: StudentCreate,
    db: Session = Depends(get_db)
):
    student = db.get(Student, student_id)

    if student is None:
        raise HTTPException(
            status_code=404,
            detail="Student not found"
        )

    student.Name = student_data.Name
    student.Age = student_data.Age
    student.Gender = student_data.Gender

    try:
        db.commit()
        db.refresh(student)
        return student

    except Exception:
        db.rollback()
        raise HTTPException(
            status_code=500,
            detail="Unable to update student"
        )

Notice there is no UPDATE statement and no db.add(). The session is tracking the object it loaded, so when you change its attributes and call commit(), SQLAlchemy generates UPDATE Students SET ... WHERE Id = ? for you. This is the same change tracking EF Core does. The same StudentCreate schema is reused, so updates get exactly the same validation as inserts.

5. Delete a student: DELETE /students/{id}

@app.delete("/students/{student_id}")
def delete_student(student_id: int, db: Session = Depends(get_db)):
    student = db.get(Student, student_id)

    if student is None:
        raise HTTPException(
            status_code=404,
            detail="Student not found"
        )

    try:
        db.delete(student)
        db.commit()

        return {
            "message": "Student deleted successfully",
            "student_id": student_id
        }

    except Exception:
        db.rollback()
        raise HTTPException(
            status_code=500,
            detail="Unable to delete student"
        )

What Happens During One Request

Follow a PUT /students/1 call from start to finish:

1. Uvicorn receives the HTTP request and passes it to FastAPI.
2. FastAPI matches the route and converts "1" to student_id: int.
3. The JSON body is validated against StudentCreate. If it fails, a 422 is returned right here.
4. Depends(get_db) opens a session from the pool and passes it in as db.
5. Your function loads the student, changes it and commits. Missing student: 404. Database error: rollback and 500.
6. The returned Student object is converted to JSON through StudentResponse.
7. The finally block in get_db() closes the session and the connection goes back to the pool.

Run and Test

uvicorn main:app --reload

Open http://127.0.0.1:8000/docs. Swagger now shows the request schema with its rules and the exact response shape of every endpoint, because they come from your Pydantic classes.

Create a student

Request: POST /students

{
  "Name": "Divyani",
  "Age": 23,
  "Gender": "Female"
}

Response 200 OK, with the Id generated by SQL Server:

{
  "Id": 1,
  "Name": "Divyani",
  "Age": 23,
  "Gender": "Female"
}

Send invalid data

Request with a one-letter name and an age of 150:

{
  "Name": "D",
  "Age": 150,
  "Gender": "Female"
}

Response 422 Unprocessable Entity. You didn't write a single if for this; Pydantic produced it from the Field rules:

{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "Name"],
      "msg": "String should have at least 2 characters",
      "input": "D",
      "ctx": { "min_length": 2 }
    },
    {
      "type": "less_than_equal",
      "loc": ["body", "Age"],
      "msg": "Input should be less than or equal to 120",
      "input": 150,
      "ctx": { "le": 120 }
    }
  ]
}

Ask for a student that doesn't exist

Request: GET /students/999. Response 404 Not Found:

{
  "detail": "Student not found"
}

Basic vs Advanced: What Changed

Taking It Further

The API is solid now. These small changes take it closer to a production service:

• Return 201 Created from POST: @app.post("/students", response_model=StudentResponse, status_code=201), and 204 No Content from DELETE.
• Catch SQLAlchemyError (from sqlalchemy.exc) instead of Exception, and log the real error with the logging module before returning the 500.
• Move the connection string to an environment variable.
• When the project grows, split endpoints into modules with APIRouter, the FastAPI version of separate controllers.
• Add a StudentUpdate schema with optional fields and a PATCH endpoint for partial updates.

Previous: Basic CRUD in FastAPI