Связываем маршруты с действиями в базе данных: архитектура

Что мы изучим?#

Теперь мы соединяем всё: сервер из темы бэкенда и FastAPI и базу из темы баз данных. Этот урок сосредоточен на архитектуре, то есть на том, как соединяются части, и реализует первые два Endpoint-а (чтение). В следующем уроке — остальное.

Зачем это нужно?#

В теме бэкенда и FastAPI сервер хранил студентов в RAM, и они исчезали при каждом перезапуске. Теперь сервер будет сохранять в базу данных: данные, которые остаются, с ограничениями и транзакциями. Это базовая архитектура почти любого бэкенда в мире: Client → API → Database.

Объяснение с основ#

Два файла, две ответственности#

students-api/
├── main.py     ← the API layer: routes, HTTP, validation, status codes
└── db.py       ← the data layer: SQL, connections, commit (from the databases topic)

MySQL in Docker (kodkod-mysql)  ← where the data lives

Это то самое «базовое разделение кода между маршрутом и кодом доступа к базе данных», которого требует программа курса. И почему это хорошо?

• Каждый файл маленький и сфокусированный (SRP, тема чистого кода).
• Каждый слой проверяется сам по себе: db.py мы уже проверили в предыдущем уроке, без сервера.
• Один слой заменяется, не трогая другой: переход на PostgreSQL изменит только db.py. Другой интерфейс (приложение командной строки) будет использовать тот же db.py.

Таблица соответствий#

Каждый маршрут делает три вещи: (1) получает и проверяет ввод (FastAPI + проверка), (2) вызывает одну функцию в db.py, (3) переводит результат в HTTP.

Один запрос от начала до конца#

GET /students/2
   │
   ▼  Uvicorn receives the HTTP request (port 8000)
   ▼  FastAPI finds the route, converts "2" → 2 (int), calls get_student(2)
   ▼  main.get_student  →  db.get_student(2)
   ▼  db: open connection → SELECT ... WHERE id = %s (2) → fetchone → dict → close
   ▼  main: dict is not None → return it
   ▼  FastAPI: dict → JSON, status 200
   ▼  Uvicorn sends the HTTP response
{"id": 2, "name": "Avi", "age": 31, "email": "avi@mail.com"}

Весь курс на одном пути: HTTP и порт (тема о компьютере, операционной системе и сети), функции, dict и JSON (темы Python), классы и Pydantic (тема ООП), SQL (тема баз данных).

Когда создаётся таблица?#

db.init_db() вызывается один раз при запуске сервера (на верхнем уровне main.py). Благодаря IF NOT EXISTS это безопасно при каждом запуске: в первый раз таблица создаётся, а в остальные ничего не происходит.

Синтаксис#

from fastapi import FastAPI, HTTPException

import db                                   # our data layer (same folder)

app = FastAPI(title="Students API")
db.init_db()                                # create the table once, on startup

@app.get("/students/{student_id}")
def get_student(student_id: int):
    student = db.get_student(student_id)    # 1 call to the data layer
    if student is None:                     # translate "not found" to HTTP
        raise HTTPException(status_code=404, detail="student not found")
    return student                          # dict → JSON (200)

Одинаковые имена в двух слоях: функция в main.py называется get_student, как и в db.py. Столкновения нет, потому что функцию базы вызывают через имя модуля: db.get_student(...) (вспомните модули!).

Первый пример кода#

Настройка проекта (контейнер MySQL должен работать — docker start kodkod-mysql):

mkdir ~\dev\kodkod6\week-4\students-api
cd ~\dev\kodkod6\week-4\students-api
python -m venv .venv
.venv\Scripts\Activate.ps1
pip install "fastapi[standard]" mysql-connector-python
git init

.gitignore:

.venv/
__pycache__/
.env

Скопируйте db.py из темы баз данных (полный CRUD) в папку. А main.py на этом этапе — только два Endpoint-а:

from fastapi import FastAPI, HTTPException

import db

app = FastAPI(title="Students API")
db.init_db()

@app.get("/students")
def list_students():
    return db.get_all_students()

@app.get("/students/{student_id}")
def get_student(student_id: int):
    student = db.get_student(student_id)
    if student is None:
        raise HTTPException(status_code=404, detail="student not found")
    return student

Добавляем данные для проверки (база пуста! POST ещё нет). seed.py:

import db

db.init_db()
for name, age, email in [("Dana", 24, "dana@mail.com"), ("Avi", 31, "avi@mail.com")]:
    try:
        print(db.create_student(name, age, email))
    except ValueError as error:
        print("Skipped:", error)

python seed.py
uvicorn main:app --reload

В браузере:

• http://127.0.0.1:8000/students
• http://127.0.0.1:8000/students/2
• http://127.0.0.1:8000/students/99

Ответы:

[{"id":1,"name":"Dana","age":24,"email":"dana@mail.com"},{"id":2,"name":"Avi","age":31,"email":"avi@mail.com"}]

{"id":2,"name":"Avi","age":31,"email":"avi@mail.com"}

{"detail":"student not found"}

Разбор кода построчно#

А теперь важный эксперимент: остановите сервер (Ctrl + C) и запустите снова. Обновите /students. Данные на месте! В отличие от темы FastAPI, они в базе — на сервере MySQL в контейнере, который записывает их на накопитель (тема о компьютере!). Их можно увидеть и в Database Client.

Ещё один пример#

Ещё один Endpoint, который использует db.py с параметром запроса: поиск. Добавьте в db.py функцию search_students из практики предыдущего урока, а в main.py:

@app.get("/students/search")
def search_students(name: str):
    if len(name.strip()) < 2:
        raise HTTPException(status_code=400, detail="search text must have at least 2 characters")
    return db.search_students(name)

Где в файле? /students/search — фиксированный путь, поэтому он должен идти до /students/{student_id} (тема FastAPI, ошибка 3!). Иначе search будет пойман как student_id → 422.

GET /students/search?name=av   → 200 [{"id": 2, "name": "Avi", ...}]
GET /students/search?name=a    → 400 {"detail": "search text must have at least 2 characters"}
GET /students/search           → 422 (name is required)

Обратите внимание на разделение работы: проверка (не меньше 2 символов) — в main.py, потому что это правило API. SQL (LIKE, lower) — в db.py.

Типичная ошибка#

Ошибка 1: SQL внутри маршрута.

@app.get("/students/{student_id}")
def get_student(student_id: int):
    connection = mysql.connector.connect(
        host="localhost", user="root", password="kodkod", database="school")
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM students WHERE id = %s", (student_id,))
    ...

Это «работает», но теперь SQL в двух местах, данные для подключения в двух местах, и маршрут, который делает всё. Маршрут = ввод + вызов db + перевод.

Ошибка 2: вернуть None из маршрута.

@app.get("/students/{student_id}")
def get_student(student_id: int):
    return db.get_student(student_id)

Для несуществующего студента возвращается 200 с null. Клиент подумает, что всё удалось. «Не найдено» = 404.

Ошибка 3: сервер не запускается — контейнер MySQL не работает. db.init_db() выполняется, когда Uvicorn загружает main.py, поэтому сервер падает уже при запуске:

mysql.connector.errors.InterfaceError: 2003: Can't connect to MySQL server on 'localhost:3306' ...

Проверьте через docker ps и запустите: docker start kodkod-mysql (тема баз данных!).

Ошибка 4: ModuleNotFoundError: No module named 'db'. db.py не в той же папке, что main.py, или вы запускаете uvicorn из другой папки. Запускайте из папки проекта.

Практика#

Упражнения опираются только на то, что пройдено в этом уроке и в предыдущих. Сначала попробуйте сами, и только потом открывайте решение.

Упражнение 1: Два слоя#

Для каждой строки кода — к чему она относится: к main.py (слой HTTP) или к db.py (слой данных)?

1. cursor.execute("SELECT * FROM students WHERE id = %s", (student_id,))
2. raise HTTPException(status_code=404, detail="student not found")
3. @app.get("/students/{student_id}")
4. mysql.connector.connect(host="localhost", user="root", ...)
5. return db.get_student(student_id)

Упражнение 2: Маршруты для чтения#

У вас есть db.py из предыдущего урока (с init_db, get_all_students и get_student). Напишите main.py с:

• db.init_db() при запуске сервера;
• GET /students — все студенты;
• GET /students/{student_id} — один студент или 404 с "student not found".

from fastapi import FastAPI, HTTPException

import db

app = FastAPI(title="Students API")
db.init_db()


@app.get("/students")
def list_students():
    return db.get_all_students()


@app.get("/students/{student_id}")
def get_student(student_id: int):
    student = db.get_student(student_id)
    if student is None:
        raise HTTPException(status_code=404, detail="student not found")
    return student

Упражнение 3: Один запрос — от начала до конца#

Расставьте по порядку то, что происходит в запросе GET /students/2:

• (а) db.get_student(2) выполняет SELECT ... WHERE id = %s с (2,)
• (б) FastAPI преобразует "2" из пути в int
• (в) Ответ возвращается в виде JSON с кодом 200
• (г) Вызывается main.get_student(2)
• (д) Строка возвращается в виде словаря

GET /students/2
  -> FastAPI: student_id = 2 (int)
  -> main.get_student(2)
  -> db.get_student(2)
  -> SELECT ... WHERE id = %s   (2,)
  -> {'id': 2, 'name': 'Avi', ...}
  -> JSON, 200

Упражнение 4: Endpoint поиска#

Добавьте в main.py маршрут GET /students/search?name=..., который использует db.search_students. Имя короче 2 символов → 400. Где нужно определить маршрут и почему?

@app.get("/students/search")
def search_students(name: str):
    if len(name.strip()) < 2:
        raise HTTPException(status_code=400, detail="too short")
    return db.search_students(name)

Упражнение 5: Что не так?#

@app.get("/students/{student_id}")
def get_student(student_id: int):
    connection = mysql.connector.connect(
        host="localhost", user="root", password="kodkod", database="school")
    cursor = connection.cursor()
    cursor.execute("SELECT * FROM students WHERE id = %s", (student_id,))
    return cursor.fetchone()

Найдите в этом коде три проблемы.

Итоги#

• main.py = слой API: маршруты, проверка и коды состояния. db.py = слой данных: SQL, соединения и commit. Никакого SQL в маршрутах.
• Соответствия: GET→SELECT, POST→INSERT (201), PUT→UPDATE, DELETE→DELETE (204). None/False → 404. ValueError → 409.
• Маршрут = ввод → один вызов db → перевод в HTTP.
• db.init_db() — один раз при запуске сервера (IF NOT EXISTS).
• Фиксированные пути (/students/search) — до /students/{student_id}.
• Данные переживают перезапуск сервера, потому что они в базе.

Официальная документация#

• FastAPI — Bigger Applications
• FastAPI — Handling Errors