Python 12 - Moving to PostgreSQL: SQLAlchemy, Sessions, and Schemas

Replace the JSON file with a real database, connect to it safely through .env, define the table in Python, and hand every request its own database session.

Where StudyDesk is now

src/studydesk/
├── __init__.py          ← full CRUD on /students, CORS
├── models.py            ← Pydantic models
├── utils.py             ← JSON file helpers
└── data/students.json

By the end of this article, StudyDesk is connected to PostgreSQL, has a students table defined in SQLAlchemy, loads settings from .env, gives each request a database session through dependency injection, and has Pydantic schemas ready for the database-backed routes we write in Article 8.

---

Part 1: Two Clerks, One File

Ritu and her colleague Pavan both use the API at the same moment. Ritu updates student 4's marks. Pavan adds a new student. Both requests read students.json, change their copy, and write it back. Pavan's write lands last and overwrites Ritu's change. Student 4's marks silently go back to the old value.

Nobody did anything wrong. The storage is wrong. A JSON file has:

• no way to let two writers work safely at the same time
• no way to look up one record without reading all of them
• no rule that says "email must be unique" or "marks must be a number"
• no way to grow past a few thousand records

Neha says, "This is what databases are for. We move StudyDesk to PostgreSQL. And we do it so that the API's URLs, bodies, and responses do not change at all. Karan's frontend will not notice."

---

Part 2: The Pieces

Four new words. Keep this table nearby; everything in this article is one of these.

Piece	What it is	Role in StudyDesk
PostgreSQL	A database server. Stores tables of rows, enforces rules, handles many users at once.	The new home of the students
SQLAlchemy	A Python library for talking to databases. We use its ORM: tables as Python classes, rows as Python objects.	Replaces utils.py
psycopg	The driver. The low-level piece that actually speaks to PostgreSQL over the network. SQLAlchemy uses it; you rarely touch it.	Installed, then forgotten
Pydantic schemas	The same Pydantic models you know, now used only for the API boundary: what comes in and what goes out.	schemas.py

Two kinds of "model" now exist, and the names matter:

• SQLAlchemy model = a database table (models.py)
• Pydantic schema = the shape of request and response JSON (schemas.py)

The course calls them exactly this, and so will we from here on.

---

Part 3: Installing PostgreSQL

You need a PostgreSQL server running on your machine. Pick one option.

Option A: install it directly

• Windows: download the installer from postgresql.org/download/windows, run it, remember the password you set for the postgres user. Keep port 5432.
• macOS: brew install postgresql@16 then brew services start postgresql@16.
• Ubuntu/Debian: sudo apt install postgresql then sudo systemctl start postgresql.

Then create the database. Open the PostgreSQL shell (psql -U postgres on Windows/Mac, sudo -u postgres psql on Linux) and run:

CREATE DATABASE studydesk;
\q

Option B: run it in Docker (if you already have Docker)

One command, nothing else to install:

docker run --name studydesk-db \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=studydesk \
  -p 5432:5432 \
  -d postgres:16

This starts a PostgreSQL 16 server with user postgres, password postgres, and a database called studydesk, reachable on port 5432. (We go deep into Docker in Article 9. For now, this is just a quick way to get a database.)

Check it

Either way, you should now be able to connect. A quick test from the terminal:

psql "postgresql://postgres:postgres@localhost:5432/studydesk" -c "SELECT 1;"

(Replace the password if you chose a different one.) If you see a 1, the database is ready.

---

Part 4: Installing the Python Packages

uv add sqlalchemy "psycopg[binary]" python-dotenv

• sqlalchemy — the ORM
• psycopg[binary] — the PostgreSQL driver (version 3, with a pre-built binary so nothing needs compiling)
• python-dotenv — reads .env, same as in the Python series

---

Part 5: Database URL, .env, and Config

The database URL

SQLAlchemy connects using a single URL:

postgresql+psycopg://postgres:postgres@localhost:5432/studydesk
└─────┬────┘ └──┬──┘  └──┬───┘ └──┬───┘ └───┬───┘ └┬─┘ └───┬───┘
   database   driver   user   password   host    port  database name

It contains a password. From the Python series: secrets never go in code. It goes in .env.

The files

.env (project root; never committed)

DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/studydesk
ALLOWED_ORIGINS=http://127.0.0.1:5500,http://localhost:5500,http://127.0.0.1:3000,http://localhost:3000

.env.example (project root; committed)

DATABASE_URL=postgresql+psycopg://USER:PASSWORD@localhost:5432/studydesk
ALLOWED_ORIGINS=http://127.0.0.1:5500,http://localhost:5500

Check that .gitignore contains .env (uv's default file does). If not, add it.

A config module

One place that reads .env and exposes settings. Everything else imports from here.

src/studydesk/config.py

import os
from pathlib import Path

from dotenv import load_dotenv

BASE_DIR = Path(__file__).resolve().parent.parent.parent  # project root
load_dotenv(BASE_DIR / ".env")

DATABASE_URL: str = os.environ["DATABASE_URL"]

ALLOWED_ORIGINS: list[str] = [
    origin.strip()
    for origin in os.getenv("ALLOWED_ORIGINS", "").split(",")
    if origin.strip()
]

Reading it

BASE_DIR config.py → studydesk/ → src/ → project root. Three .parent calls. .resolve() turns it into an absolute path first.

os.environ["DATABASE_URL"] Square brackets, not .getenv(): if the key is missing, the app crashes at import time with a clear KeyError. A missing database URL should never be silently None.

ALLOWED_ORIGINS A comma-separated string in .env becomes a list. .strip() removes stray spaces; the if drops empty entries. The CORS list from Article 6 now lives in .env, so it can differ between your laptop and production without code changes.

---

Part 6: Creating the Engine, Session, and Base

src/studydesk/database.py

from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

from .config import DATABASE_URL

engine = create_engine(DATABASE_URL, echo=False)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


class Base(DeclarativeBase):
    """Parent class for every SQLAlchemy model."""


def get_db() -> Generator[Session, None, None]:
    """Open a database session for one request, and always close it."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

This small file has three objects and one function. Each deserves a clear picture.

The engine

engine = create_engine(DATABASE_URL, echo=False)

The engine is the connection manager. It knows the URL, opens connections to PostgreSQL, and keeps a small pool of them ready so each request does not pay the cost of a fresh network connection. There is exactly one engine for the whole app.

echo=True prints every SQL statement SQLAlchemy sends. Switch it on when you are curious or debugging; keep it off otherwise.

The session factory

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

A session is a short-lived workspace for talking to the database: you query through it, add objects to it, and finally commit (save) or roll back (discard). Think of it as a shopping basket; nothing is bought until you check out.

sessionmaker() builds a factory: calling SessionLocal() gives a new session. autocommit=False means nothing is saved until we say db.commit(). autoflush=False keeps SQLAlchemy from sending partial changes early. These are the standard FastAPI settings.

The Base class

class Base(DeclarativeBase):
    ...

Every table model will inherit from Base. Through it, SQLAlchemy keeps a registry of all tables, which is what lets us create them all with one command later.

get_db() — the session per request

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

This function looks almost normal, but it uses yield instead of return. That one word makes it a generator, and generators were not covered in the Python series, so let's look at them properly.

A normal function runs once and is done

def give_number():
    return 5

x = give_number()
print(x)   # 5

return sends a value back and the function is finished. Its local variables are thrown away. If there was code after return, it never runs.

A generator can pause and continue

Replace return with yield:

def give_numbers():
    print("start")
    yield 1
    print("middle")
    yield 2
    print("end")

Calling give_numbers() does not run the function. It returns a generator object, which is a paused function waiting to be stepped through. You step it with next():

gen = give_numbers()

print(next(gen))
# start
# 1

print(next(gen))
# middle
# 2

next(gen)
# end
# StopIteration

Read what happened:

1. First next(): the function runs until it hits yield 1. It hands out 1 and pauses right there, keeping all its local variables.
2. Second next(): the function continues from where it paused, prints "middle", hits yield 2, hands out 2, pauses again.
3. Third next(): continues, prints "end", reaches the end of the function. There is nothing more to yield, so Python raises StopIteration to say "finished".

The key idea: yield pauses the function, and the next call resumes it from the same spot. A return ends things; a yield is a bookmark.

You have already used generators without noticing. range(5) in a for loop hands out numbers one by one instead of building a list. The generator expression max((s["id"] for s in students), default=0)was the same idea.

Now read get_db() again

def get_db():
    db = SessionLocal()      # 1. runs before yield
    try:
        yield db             # 2. hand the session out, then PAUSE here
    finally:
        db.close()           # 3. runs when resumed, after the pause

Think of it as three stages:

Stage	Code	When
Setup	db = SessionLocal()	When someone first steps into the generator
Hand-over	yield db	The session is given to whoever asked; the function pauses here and waits
Cleanup	db.close()	When the generator is resumed, after the caller is done

The try / finally wraps the pause. From the Python series: finally always runs, even if an error happens. So if the route using the session crashes, the generator is still resumed, finally still runs, and the session is still closed. No leaked connections.

Why a generator instead of two functions?

We could have written open_db() and close_db() and asked every route to call both. But then one forgotten close_db() leaks a connection forever. With a generator, setup and cleanup live in one function, on either side of a single yield, and they cannot be separated. The caller simply uses what was yielded and then lets the generator finish.

This is exactly why FastAPI chose this pattern for dependencies. In Part 8 you will see FastAPI step into get_db() before the route (running the setup), take the yielded session, run the route, and then step into get_db() once more after the response is sent (running the cleanup). You never call next() yourself; FastAPI does it.

The return type hint

from collections.abc import Generator

def get_db() -> Generator[Session, None, None]:

This hint says: "this is a generator that yields Session objects." The two Nones describe advanced features (sending values in, and a final return value) that we do not use. The hint is optional, but it makes VS Code understand that db inside a route is a Session, so autocomplete works.

---

Part 7: Defining the Table with DeclarativeBase

Until now, models.py held Pydantic models. From now on, models.py holds SQLAlchemy tables and the Pydantic models move to schemas.py (next part). Replace models.py entirely.

src/studydesk/models.py

from sqlalchemy import Integer, String
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(50), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    age: Mapped[int] = mapped_column(Integer, nullable=False)
    city: Mapped[str] = mapped_column(String(50), nullable=False)
    marks: Mapped[int] = mapped_column(Integer, nullable=False)

    def __repr__(self) -> str:
        return f"<Student id={self.id} name={self.name!r} marks={self.marks}>"

Reading it

class Student(Base) A Python class that is a table. One instance = one row.

__tablename__ = "students" The actual table name in PostgreSQL. Convention: plural, lowercase.

id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) Each line is a column. Mapped[int] is the Python type of the attribute. mapped_column(...) describes the database column:

Option	Meaning
Integer, String(50)	Column type and max length
primary_key=True	Unique identifier for the row
autoincrement=True	PostgreSQL assigns 1, 2, 3… — we no longer compute max(id) + 1 ourselves
nullable=False	Required; the database rejects a row without it
unique=True	No two rows may share this value — our email check, now enforced by the database
index=True	Build an index so lookups by email are fast

No grade column Grade stays computed. It will be calculated in the response schema, exactly like Article 4.

__repr__ Only for nicer printing while debugging. !r shows the name in quotes.

Pydantic rules vs database rules

Notice the database knows less than Pydantic: it does not know marks must be 0–100, or that names must not contain digits. That is fine. The split is:

• Pydantic schemas validate what clients send (detailed rules, friendly 422 messages).
• The database enforces structural truths that must hold no matter which code writes to it (types, required, unique).

Both layers together are safer than either alone.

---

Part 8: Managing Sessions with get_db and Dependency Injection

The problem

Every database route needs a session. Without help, each route would look like this:

@app.get("/students")
def get_students():
    db = SessionLocal()
    try:
        ...  # do the work
    finally:
        db.close()

Ten routes, ten copies of the same opening and closing. The first time someone forgets the finally, a connection leaks on every request until the pool runs dry and the server stops answering.

FastAPI's answer is dependency injection. The idea in plain words: a route says what it needs, and FastAPI provides it. The route never builds the thing itself.

What is a dependency?

A dependency is just a function that FastAPI calls for you before running your route, and whose result it passes into the route as a parameter.

Start with the simplest possible example, with no database at all:

from fastapi import Depends

def get_greeting():
    return "Hello from a dependency"

@app.get("/hello")
def hello(greeting: str = Depends(get_greeting)):
    return {"message": greeting}

Calling GET /hello returns {"message": "Hello from a dependency"}. What happened:

1. FastAPI sees the parameter greeting with Depends(get_greeting).
2. It calls get_greeting() itself.
3. It passes the return value in as greeting.
4. It runs hello().

Depends(get_greeting) does not call the function. It hands FastAPI the function and says "call this when you need the value." That is why there are no brackets after get_greeting inside Depends(...).

Dependencies can be generators

Now the real one. get_db() is a generator, so FastAPI treats it in two halves:

@app.get("/db-check")
def db_check(db: Session = Depends(get_db)):
    ...

1. Before running db_check, FastAPI steps into get_db(). It runs up to yield: the session is opened.
2. The yielded session is passed in as db.
3. db_check runs.
4. After the response is sent, FastAPI steps into get_db() again. Execution continues after yield, so finally runs and the session is closed.

This happens for every request, separately. Two clerks hitting the API at the same moment each get their own session. And because cleanup is inside the generator, no route can forget it.

Dependencies can depend on other dependencies

This is where Depends becomes powerful.

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    ...  # look up the user from the token

get_current_user itself uses Depends twice. A route that writes user: User = Depends(get_current_user) triggers a chain: FastAPI opens the session, reads the token, looks up the user, and hands the route a ready User object. The route only knows it received a user. That chaining is why FastAPI calls it a dependency tree.

What Annotated really is

So far we have written Depends with =:

db: Session = Depends(get_db)

This works, but it is a small lie: it uses the default value slot to carry something that is not a default value. The modern way uses Annotated.

Annotated comes from Python's typing module. Its job is simple:

Annotated[<the real type>, <extra information>, <more extra information>, ...]

To Python itself, Annotated[Session, Depends(get_db)] means exactly Session. Python ignores everything after the first item. VS Code sees a Session and autocompletes db.query, db.commit, and so on.

The extra items are metadata: notes attached to the type that other tools can read. FastAPI is one of those tools. When it inspects a route's parameters, it looks inside each Annotated and asks: "Is there a Path(...) here? A Query(...)? A Depends(...)?" and acts on what it finds.

So all three of these are the same mechanism:

Parameter	Python sees	FastAPI reads the metadata and…
student_id: Annotated[int, Path(gt=0)]	int	takes it from the URL path, validates > 0
limit: Annotated[int, Query(le=100)] = 10	int	takes it from ?limit=, validates ≤ 100
db: Annotated[Session, Depends(get_db)]	Session	calls get_db() and injects the result

The first item is for Python and your editor. The second item is an instruction for FastAPI. That is the whole idea of Annotated.

Why the Annotated form is better than = Depends(...)

• The parameter has no fake default, so Python's rules about ordering parameters are not disturbed.
• The type and its FastAPI instruction travel together, so the combination can be reused.

That reuse is the next step.

DbSession: define it once

DbSession = Annotated[Session, Depends(get_db)]

This is a plain variable assignment. DbSession is now a name for the whole Annotated[...] expression. Anywhere you write

def db_check(db: DbSession):

Python expands it to db: Annotated[Session, Depends(get_db)]. Same type, same metadata, same behaviour, without repeating Depends(get_db) in every route. Change the alias in one place and every route follows.

You will see this pattern throughout the rest of the series:

DbSession = Annotated[Session, Depends(get_db)]           
CurrentUser = Annotated[User, Depends(get_current_user)] 
AdminUser = Annotated[User, Depends(require_admin)]  

Each one reads like a type and acts like a dependency.

Why this matters beyond convenience

Because the route depends on a session, not on how to make one, the session can be swapped without touching the route.

app.dependency_overrides[get_db] = get_test_db

One line, and every route that says db: DbSession now receives a session to a test database. The routes never find out. That is dependency injection paying for itself.

---

Part 9: Pydantic Input and Output Schemas

Create schemas.py. It is the previous article Pydantic code with one important addition for reading from the database.

src/studydesk/schemas.py

from pydantic import BaseModel, ConfigDict, EmailStr, Field, computed_field, field_validator


def clean_text(value: str) -> str:
    """Strip spaces, reject digits, convert to Title Case."""
    value = value.strip()
    if any(char.isdigit() for char in value):
        raise ValueError("must not contain digits")
    return value.title()


class StudentBase(BaseModel):
    """Fields shared by every student shape."""

    name: str = Field(min_length=2, max_length=50, description="Full name", examples=["Priya Sharma"])
    email: EmailStr = Field(description="Email address", examples=["priya@example.com"])
    age: int = Field(ge=5, le=100, description="Age in years", examples=[17])
    city: str = Field(min_length=2, max_length=50, description="City", examples=["Ahmedabad"])
    marks: int = Field(ge=0, le=100, description="Marks out of 100", examples=[88])

    @field_validator("name", "city")
    @classmethod
    def validate_text(cls, value: str) -> str:
        return clean_text(value)


class StudentCreate(StudentBase):
    """Input schema for POST /students."""


class StudentUpdate(BaseModel):
    """Input schema for PUT /students/{id}. Every field is optional."""

    name: str | None = Field(default=None, min_length=2, max_length=50, examples=["Priya Sharma"])
    email: EmailStr | None = Field(default=None, examples=["priya@example.com"])
    age: int | None = Field(default=None, ge=5, le=100, examples=[17])
    city: str | None = Field(default=None, min_length=2, max_length=50, examples=["Ahmedabad"])
    marks: int | None = Field(default=None, ge=0, le=100, examples=[88])

    @field_validator("name", "city")
    @classmethod
    def validate_text(cls, value: str | None) -> str | None:
        if value is None:
            return value
        return clean_text(value)


class StudentOut(StudentBase):
    """Output schema: what every student response looks like."""

    model_config = ConfigDict(from_attributes=True)

    id: int = Field(gt=0, description="Unique student ID", examples=[1])

    @computed_field
    @property
    def grade(self) -> str:
        """Letter grade calculated from marks."""
        if self.marks >= 85:
            return "A"
        if self.marks >= 70:
            return "B"
        if self.marks >= 50:
            return "C"
        return "D"


class StudentListOut(BaseModel):
    """Output schema for GET /students."""

    total: int
    total_pages: int
    offset: int
    limit: int
    count: int
    students: list[StudentOut]

What changed from Article 5

Student is now StudentOut The name says what it is: the shape of a student going out to the client. The SQLAlchemy class owns the name Student now, and two classes with the same name in one project is a recipe for confusion.

model_config = ConfigDict(from_attributes=True) This is the key line. Until now, Pydantic models were built from dicts: Student(**record). A SQLAlchemy row is not a dict; it is an object with attributes (row.name, row.marks). from_attributes=True tells Pydantic "you may also build this schema by reading attributes from an object." With it, FastAPI can take a database row and produce StudentOut JSON directly. Without it, you get a validation error saying it expected a dict.

grade is still computed From marks, in the output schema. The database does not store it, and the client still sees it. Nothing about the response changed.

StudentListOut The paginated response from Article 3, now written down as a schema. In future articles it becomes the response_model of GET /students, which means it also appears fully in /docs.

Input vs output

Direction	Schemas	Built from
In (request body)	StudentCreate, StudentUpdate	JSON from the client
Out (response)	StudentOut, StudentListOut	SQLAlchemy rows, via from_attributes

The same validation rules protect the way in. The same grade calculation decorates the way out. The storage in the middle is now a real database.

---

Part 10: Creating the Table and Seeding It

Creating tables at startup

SQLAlchemy can create every table registered on Base with one call: Base.metadata.create_all(engine). The right moment is when the app starts. FastAPI's lifespan hook runs code once at startup and once at shutdown.

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)   # startup
    yield
    # shutdown (nothing yet)

app = FastAPI(lifespan=lifespan, ...)

Same yield pattern as get_db: before it is startup, after it is shutdown. create_all is safe to run every time: it only creates tables that do not exist yet. (It will not change existing tables; that is what Alembic)

Moving the JSON students into the database

A one-time script that reads students.json and inserts each record. This is the last time the JSON file is used.

src/studydesk/seed.py

import json
from pathlib import Path

from .database import Base, SessionLocal, engine
from .models import Student

DATA_FILE = Path(__file__).parent / "data" / "students.json"


def seed() -> None:
    Base.metadata.create_all(bind=engine)

    with open(DATA_FILE, "r", encoding="utf-8") as f:
        records = json.load(f)

    db = SessionLocal()
    try:
        if db.query(Student).count() > 0:
            print("Table already has data, skipping.")
            return

        for record in records:
            record.pop("grade", None)
            db.add(Student(**record))
        db.commit()
        print(f"Inserted {len(records)} students.")
    finally:
        db.close()


if __name__ == "__main__":
    seed()

Reading it

db.query(Student).count() Ask the database how many rows the students table has. If it is not empty, do nothing; running the seed twice must not duplicate everyone.

db.add(Student(**record)) Build a row object from the dict (the ** unpacking from Article 4) and put it in the session's basket. record.pop("grade", None) is only there for safety in case anyone left grade in the file.

db.commit() Check out the basket: all ten INSERTs are sent and saved together. Until this line, nothing is in the database.

if __name__ == "__main__": From the Python series: run seed() only when this file is executed directly.

Run it:

uv run python -m studydesk.seed

Inserted 10 students.

-m studydesk.seed runs the module by its package path, which makes the relative imports (.database, .models) work. Run it again and you get "Table already has data, skipping."

Check in psql:

SELECT id, name, marks FROM students ORDER BY id;

Ten rows, IDs 1 to 10, assigned by PostgreSQL.

---

Part 11: Wiring It Into the App

The existing CRUD routes still read the JSON file; we rewrite them one by one in future article. Today, __init__.py gets:

1. the lifespan that creates tables,
2. CORS origins from config,
3. one new route, GET /db-check, that proves dependency injection and the database work together.

The JSON routes import from schemas now (since models.py changed meaning), and utils.py is updated to match.

src/studydesk/utils.py

import json
from pathlib import Path

from .schemas import StudentOut

DATA_FILE = Path(__file__).parent / "data" / "students.json"


def load_students() -> list[dict]:
    """Read all students from the JSON file, validated through the StudentOut schema."""
    with open(DATA_FILE, "r", encoding="utf-8") as f:
        raw_records = json.load(f)
    return [StudentOut(**record).model_dump() for record in raw_records]


def save_students(students: list[dict]) -> None:
    """Validate every record and write the list back to the JSON file."""
    clean_records = [StudentOut(**record).model_dump(exclude={"grade"}) for record in students]
    with open(DATA_FILE, "w", encoding="utf-8") as f:
        json.dump(clean_records, f, indent=2)


def find_student(students: list[dict], student_id: int) -> dict | None:
    """Return the student dict with the given id, or None."""
    for student in students:
        if student["id"] == student_id:
            return student
    return None


def email_exists(students: list[dict], email: str, exclude_id: int | None = None) -> bool:
    """True if another student already uses this email (case-insensitive)."""
    for student in students:
        if student["id"] == exclude_id:
            continue
        if student["email"].lower() == email.lower():
            return True
    return False

src/studydesk/__init__.py

from contextlib import asynccontextmanager
from typing import Annotated, Literal

import uvicorn
from fastapi import Depends, FastAPI, HTTPException, Path, Query, status
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from .config import ALLOWED_ORIGINS
from .database import Base, engine, get_db
from .models import Student
from .schemas import StudentCreate, StudentOut, StudentUpdate
from .utils import email_exists, find_student, load_students, save_students


@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)
    yield


app = FastAPI(
    title="StudyDesk API",
    description="Student records API for a coaching institute.",
    version="0.2.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

DbSession = Annotated[Session, Depends(get_db)]


@app.get("/", tags=["General"])
def home():
    return {"message": "Welcome to StudyDesk API"}


@app.get("/db-check", tags=["General"], summary="Check the database connection")
def db_check(db: DbSession):
    total = db.scalar(select(func.count()).select_from(Student))
    return {"database": "ok", "students_in_db": total}


@app.get("/students", tags=["Students"], summary="Get all students")
def get_students(
    city: Annotated[
        str | None,
        Query(description="Filter by city (case-insensitive).", examples=["Ahmedabad"]),
    ] = None,
    grade: Annotated[
        str | None,
        Query(
            description="Filter by grade: A, B, C, or D.",
            pattern="^[ABCD]$",
            examples=["A"],
        ),
    ] = None,
    min_marks: Annotated[
        int | None,
        Query(description="Only students with marks >= this value.", ge=0, le=100),
    ] = None,
    sort_by: Annotated[
        Literal["id", "name", "age", "marks"],
        Query(description="Field to sort by."),
    ] = "id",
    order: Annotated[
        Literal["asc", "desc"],
        Query(description="Sort direction: asc or desc."),
    ] = "asc",
    offset: Annotated[
        int,
        Query(description="Number of students to skip.", ge=0),
    ] = 0,
    limit: Annotated[
        int,
        Query(description="Maximum number of students to return (1-100).", ge=1, le=100),
    ] = 10,
):
    students = load_students()

    # --- filtering ---
    if city:
        students = [s for s in students if s["city"].lower() == city.lower()]

    if grade:
        students = [s for s in students if s["grade"] == grade.upper()]

    if min_marks is not None:
        students = [s for s in students if s["marks"] >= min_marks]

    # --- sorting ---
    students = sorted(students, key=lambda s: s[sort_by], reverse=(order == "desc"))

    # --- pagination ---
    total = len(students)
    total_pages = (total + limit - 1) // limit if total > 0 else 0
    page = students[offset : offset + limit]

    return {
        "total": total,
        "total_pages": total_pages,
        "offset": offset,
        "limit": limit,
        "count": len(page),
        "students": page,
    }


@app.get("/students/{student_id}", tags=["Students"], summary="Get one student by ID")
def get_student(
    student_id: Annotated[
        int,
        Path(
            title="Student ID",
            description="The unique ID of the student. Must be a positive integer.",
            gt=0,
            examples=[1],
        ),
    ],
):
    students = load_students()
    student = find_student(students, student_id)
    if student is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Student with id {student_id} not found",
        )
    return student


@app.post(
    "/students",
    tags=["Students"],
    summary="Create a new student",
    status_code=status.HTTP_201_CREATED,
)
def create_student(payload: StudentCreate):
    students = load_students()

    if email_exists(students, payload.email):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"A student with email {payload.email} already exists",
        )

    next_id = max((s["id"] for s in students), default=0) + 1
    new_student = StudentOut(id=next_id, **payload.model_dump())

    students.append(new_student.model_dump())
    save_students(students)

    return new_student


@app.put("/students/{student_id}", tags=["Students"], summary="Update a student (partial)")
def update_student(
    student_id: Annotated[int, Path(title="Student ID", gt=0, examples=[1])],
    payload: StudentUpdate,
):
    students = load_students()
    student = find_student(students, student_id)
    if student is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Student with id {student_id} not found",
        )

    changes = payload.model_dump(exclude_unset=True)
    if not changes:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="No fields provided to update",
        )

    if "email" in changes and email_exists(students, changes["email"], exclude_id=student_id):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"A student with email {changes['email']} already exists",
        )

    merged = student.copy()
    merged.update(changes)
    merged.pop("grade", None)
    updated_student = StudentOut(**merged)

    for index, s in enumerate(students):
        if s["id"] == student_id:
            students[index] = updated_student.model_dump()
            break

    save_students(students)
    return updated_student


@app.delete(
    "/students/{student_id}",
    tags=["Students"],
    summary="Delete a student",
    status_code=status.HTTP_204_NO_CONTENT,
)
def delete_student(
    student_id: Annotated[int, Path(title="Student ID", gt=0, examples=[1])],
):
    students = load_students()
    student = find_student(students, student_id)
    if student is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Student with id {student_id} not found",
        )

    students = [s for s in students if s["id"] != student_id]
    save_students(students)


def main() -> None:
    """Start the development server."""
    uvicorn.run("studydesk:app", host="127.0.0.1", port=8000, reload=True)

Reading the new parts

lifespan — creates the students table on startup if it does not exist.

ALLOWED_ORIGINS from config — the CORS list now comes from .env.

DbSession = Annotated[Session, Depends(get_db)] — defined once, used by every database route.

GET /db-check

total = db.scalar(select(func.count()).select_from(Student))

Our first real query. select(func.count()).select_from(Student) builds the SQL SELECT count(*) FROM students. db.scalar(...) runs it and returns the single value. If the database is unreachable, this route raises and you see the real error in the terminal, which makes it a useful health check.

The CRUD routes — unchanged in behaviour; only the schema name Student → StudentOut. They still read the JSON file. In Article 8, each one becomes a database route.

Run it

uv run studydesk

In the terminal you will see the usual startup lines. Open http://127.0.0.1:8000/db-check:

{"database": "ok", "students_in_db": 10}

The request came in, get_db() opened a session, the route counted rows in PostgreSQL, the response went out, and the session was closed. Dependency injection, end to end.

---

Where StudyDesk is after this article

studydesk/
├── .env                      ← DATABASE_URL, ALLOWED_ORIGINS (not committed)
├── .env.example
└── src/
    └── studydesk/
        ├── __init__.py       ← lifespan, DbSession, GET /db-check; CRUD still on JSON
        ├── config.py         ← reads .env
        ├── database.py       ← engine, SessionLocal, Base, get_db
        ├── models.py         ← SQLAlchemy Student table
        ├── schemas.py        ← Pydantic StudentCreate, StudentUpdate, StudentOut, StudentListOut
        ├── seed.py           ← one-time JSON → PostgreSQL import
        ├── utils.py          ← JSON helpers (retired in Article 8)
        └── data/students.json

PostgreSQL has a students table with ten rows. The API can reach it. The schemas are ready. Next article, the routes switch over.

---

Concept Quiz

Answers are below. No coding needed.

Q1. Name two concrete problems with using a JSON file as storage that PostgreSQL solves.

Q2. What is the difference between a SQLAlchemy model and a Pydantic schema in this project?

Q3. Why does config.py use os.environ["DATABASE_URL"] instead of os.getenv("DATABASE_URL")?

Q4. What is the engine, and how many should an app have?

Q5. What does a session do, and what does commit() mean in the shopping-basket picture?

Q6. Explain what happens in get_db() before, at, and after yield.

Q7. What does db: DbSession on a route tell FastAPI to do?

Q8. What does unique=True on the email column give us that the old email_exists() helper did not?

Q9. What does from_attributes=True allow, and what breaks without it?

Q10. Base.metadata.create_all() runs on every startup. Why is that safe, and what can it not do?

---

Answers

1. Two writers at once overwrite each other's changes (lost updates); every read loads all records; no enforced rules like unique emails; does not scale past a few thousand records.
2. A SQLAlchemy model describes a database table (models.py). A Pydantic schema describes the JSON shape at the API boundary (schemas.py). The row is the model; the request/response is the schema.
3. Square brackets raise KeyError immediately if the key is missing. A missing database URL should crash loudly at startup, not become None and fail somewhere later.
4. The connection manager that holds a pool of connections to PostgreSQL. Exactly one per app.
5. A short-lived workspace where you query and stage changes. commit() is checkout: all staged changes are written to the database together. Before that, nothing is saved.
6. Before: SessionLocal() opens a session. At: the session is handed to the route and the function pauses. After: when the route is done, execution resumes and finally closes the session, even if the route raised an error.
7. Call get_db() to obtain a session, pass it in as db, run the route, then resume get_db() to close the session. The route never opens or closes a session itself.
8. The database itself refuses a duplicate email, no matter which code inserts it, and it is checked atomically, so two simultaneous requests cannot both succeed. The Python helper could be bypassed and had a race condition.
9. It lets Pydantic build a schema by reading attributes from an object (a SQLAlchemy row) instead of keys from a dict. Without it, returning a row through StudentOut fails with a validation error expecting a dict.
10. It only creates tables that do not already exist, so repeated runs do nothing. It cannot alter existing tables (add a column, change a type); that needs migrations with Alembic.

---

Epilogue

Aman opens psql, runs SELECT * FROM students;, and sees the same ten names that used to live in a text file. Ritu and Pavan can now save at the same second; PostgreSQL will sort it out.

Neha scrolls to the CRUD routes. "These five still read the file. Let's rewrite them, one at a time, and make the responses typed with response_model so /docs shows exactly what comes back."

In the next article: database endpoints — fetch one, create, update, delete, and the full list with filters, sorting, and pagination, all through SQLAlchemy.