# 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

```plaintext
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."

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/fa5beffc-b749-4d2a-8995-9b07815c01f2.png align="center")

---

## 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:

```sql
CREATE DATABASE studydesk;
\q
```

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

One command, nothing else to install:

```bash
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:

```bash
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

```bash
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:

```plaintext
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)

```plaintext
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)

```plaintext
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`

```python
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`

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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`:

```python
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()`:

```python
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

```python
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

```python
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 `None`s 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.

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/c52926c0-cd03-47d5-b132-16fab0d5b33f.png align="center")

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/228cb936-c001-476b-b810-823dfff7225a.png align="center")

---

## 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`

```python
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.

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/f35e5b2d-3529-4511-9334-1986ed8b64bb.png align="center")

---

## 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:

```python
@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:

```python
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:

```python
@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.

```python
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 `=`:

```python
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:

```python
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

```python
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

```python
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:

```python
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.

```python
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.

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/2a9699ba-6bb9-4594-95e5-8b7e91b715a0.png align="center")

---

## 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`

```python
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.

```python
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`

```python
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 `INSERT`s 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:

```bash
uv run python -m studydesk.seed
```

```plaintext
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`:

```sql
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`

```python
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`

```python
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`

```python
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

```bash
uv run studydesk
```

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

```json
{"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.

![](https://cdn.hashnode.com/uploads/covers/68822b6ad516e7a593530cf9/43e8ee09-baa4-400d-8fbb-cc5cbc5723b4.png align="center")

---

## Where StudyDesk is after this article

```plaintext
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.