Текущий пользователь как dependency

В прошлом уроке мы научились выпускать access token: sub внутри, exp ограничивает срок, проверка подписи HS256 обнаруживает изменение payload без знания ключа. Но сам по себе выпуск токена не защищает ни один маршрут. Сейчас мы напишем зависимость, которая принимает этот токен, проверяет его и возвращает реального User из базы — и сразу применим её к POST бронирования.

get_current_user: извлечение Bearer, проверка JWT и поиск User

Скопируйте папку auth_demo из урока 31 под новым именем auth_owned_demo. Исключите из копии .venv, .env, файл базы данных (*.db) и папки __pycache__. Все остальные файлы из урока 31 остаются на месте — в следующем разделе мы заменим только models.py и routers/bookings.py, и тогда же проинициализируем новую базу.

Теперь добавьте файл app/security.py с зависимостью get_current_user:

import jwt
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import User
from app.settings import Settings, get_settings

bearer_scheme = HTTPBearer(auto_error=False)

def get_current_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
    db: Session = Depends(get_db),
    settings: Settings = Depends(get_settings),
) -> User:
    if credentials is None:
        raise HTTPException(status_code=401, detail="Not authenticated",
                            headers={"WWW-Authenticate": "Bearer"})
    try:
        payload = jwt.decode(credentials.credentials, settings.secret_key,
                             algorithms=["HS256"], options={"require": ["exp", "sub"]})
        token_subject = payload["sub"]
    except (jwt.InvalidTokenError, ValueError, TypeError, OverflowError):
        raise HTTPException(status_code=401, detail="Invalid or expired token",
                            headers={"WWW-Authenticate": "Bearer"})
    user = db.scalars(select(User).where(User.token_subject == token_subject)).first()
    if user is None:
        raise HTTPException(status_code=401, detail="User not found",
                            headers={"WWW-Authenticate": "Bearer"})
    return user

HTTPBearer и auto_error=False. HTTPBearer — схема безопасности OpenAPI. Именно она добавляет кнопку «Authorize» в Swagger: кнопка появляется потому, что схема зарегистрирована в спецификации, а не потому что ответ содержит WWW-Authenticate. Когда запрос приходит без заголовка Authorization: Bearer ..., FastAPI 0.115.6 по умолчанию возвращает 403. auto_error=False меняет это поведение: вместо автоматического отказа bearer_scheme возвращает None, и наш код явно отвечает 401 с нужным заголовком.

jwt.decode: три конкретных требования. Параметр algorithms=["HS256"] разрешает только HS256. Библиотека читает alg из заголовка токена и проверяет его по этому доверенному списку; сам список мы не берём из токена. Опция options={"require": ["exp", "sub"]} делает оба поля обязательными при проверке содержимого после подписи. Стандартная проверка срока гарантирует: просроченный exp даёт ExpiredSignatureError, который попадает в тот же except jwt.InvalidTokenError.

После успешного decode мы читаем payload["sub"]. PyJWT проверяет тип: если sub не строка, библиотека выбрасывает исключение. Значит, в except попадут все реальные проблемы с токеном — невалидная подпись, изменённый payload, просрочка, отсутствие exp/sub, нестроковый sub — и любой из этих случаев даст клиенту 401.

SELECT после проверки токена. SQL выполняется только когда подпись уже проверена и token_subject прочитан. Это разделение важно: ошибка подключения к базе или отсутствие таблицы не смешивается с ошибкой credentials. Если учётная запись с таким token_subject не найдена — ещё один явный 401. Результат зависимости — реальный User из базы, а не непроверенный payload.

Каждый ответ об ошибке несёт заголовок WWW-Authenticate: Bearer — это требование RFC 6750 для Bearer-схемы.

POST бронирования: user_id из проверенного пользователя

Замените app/models.py полным файлом ниже — Booking получает обязательную колонку user_id с внешним ключом на users.id:

from secrets import token_hex
from datetime import datetime
from sqlalchemy import ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Room(Base):
    __tablename__ = "rooms"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    capacity: Mapped[int]
    address_city: Mapped[str]
    address_street: Mapped[str]
    comment: Mapped[str | None]
    floor: Mapped[int]
    has_projector: Mapped[bool]
    internal_note: Mapped[str]
    bookings: Mapped[list["Booking"]] = relationship(back_populates="room")

class Booking(Base):
    __tablename__ = "bookings"
    id: Mapped[int] = mapped_column(primary_key=True)
    room_id: Mapped[int] = mapped_column(ForeignKey("rooms.id"))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    starts_at: Mapped[datetime] = mapped_column(DateTime(timezone=False))
    ends_at: Mapped[datetime] = mapped_column(DateTime(timezone=False))
    room: Mapped["Room"] = relationship(back_populates="bookings")

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True)
    hashed_password: Mapped[str]
    token_subject: Mapped[str] = mapped_column(unique=True, default=lambda: token_hex(32))

create_all создаёт таблицы с нуля, а не добавляет колонки в существующую таблицу. Именно поэтому мы работаем в новой папке с новой пустой базой: старая папка auth_demo остаётся нетронутой.

Теперь замените app/routers/bookings.py:

from datetime import timezone
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Room, Booking, User
from app.security import get_current_user
from app.schemas.bookings import BookingIn, BookingOut
from app.booking_utils import find_conflict, to_db_utc

router = APIRouter()

def public_booking(booking: Booking) -> BookingOut:
    return BookingOut(
        id=booking.id, room_id=booking.room_id,
        starts_at=booking.starts_at.replace(tzinfo=timezone.utc),
        ends_at=booking.ends_at.replace(tzinfo=timezone.utc),
    )

@router.post("/{room_id}/bookings", status_code=201, response_model=BookingOut) def create_booking(room_id: int, data: BookingIn, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): if db.get(Room, room_id) is None: raise HTTPException(status_code=404, detail="Room not found") if find_conflict(db, room_id, data.starts_at, data.ends_at) is not None: raise HTTPException(status_code=409, detail="Booking conflict") booking = Booking( room_id=room_id, user_id=current_user.id, starts_at=to_db_utc(data.starts_at), ends_at=to_db_utc(data.ends_at), ) db.add(booking) db.commit() db.refresh(booking) return public_booking(booking)

@router.get("/{room_id}/bookings", response_model=list[BookingOut])
def list_bookings(room_id: int, db: Session = Depends(get_db)):
    if db.get(Room, room_id) is None:
        raise HTTPException(status_code=404, detail="Room not found")
    bookings = db.scalars(select(Booking).where(Booking.room_id == room_id).order_by(Booking.id)).all()
    return [public_booking(booking) for booking in bookings]

Клиент не передаёт user_id в теле запроса — обработчик берёт его только из current_user.id. Это гарантирует, что бронь всегда принадлежит тому, чей токен прошёл проверку. UTC-преобразование, проверки 404 и 409, а также публичное чтение броней через GET остались без изменений. Защищён только POST бронирования, CRUD комнат и чтение остаются публичными.

create_booking и get_current_user оба объявляют db: Session = Depends(get_db). FastAPI кеширует результат зависимости в рамках одного запроса, поэтому обе функции получат один и тот же объект Session — не два отдельных подключения.

Запуск и ручная проверка. Остановите прежний локальный сервер сочетанием Ctrl+C в его терминале, чтобы порт 8000 был свободен. Старую папку и её базу не удаляйте. Перейдите в корень новой папки auth_owned_demo. Создайте собственную .venv и установите зависимости.

В PowerShell (Python 3.12 установлен через Python Launcher):

py -3.12 -m venv .venv
.venv\Scripts\Activate.ps1
python -m pip install -r requirements.txt

В bash:

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

Задайте случайный SECRET_KEY в переменной окружения текущего терминала. В PowerShell:

$env:SECRET_KEY = python -c "import secrets; print(secrets.token_urlsafe(32))"

В bash:

export SECRET_KEY="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"

Проект не читает .env автоматически. Сохраните ключ вне репозитория и при новом запуске задавайте то же значение: смена ключа сделает прежние JWT непроверяемыми. Не выводите ключ через API и не публикуйте его.

После замены обоих файлов создайте таблицы и запустите новый сервер из того же терминала:

python -c "from app.database import init_db; init_db()"
python -m uvicorn app.main:app --reload

Откройте http://127.0.0.1:8000/docs. Новая база пустая: сначала зарегистрируйтесь через POST /api/users, затем войдите через POST /api/auth/login. Нажмите «Authorize» и вставьте полученный access_token без префикса Bearer. Создайте комнату через POST /api/rooms и подставьте её фактический id в POST /api/rooms/{room_id}/bookings. Передайте starts_at и ends_at с часовым поясом. Ожидайте 201, затем выполните GET того же пути и убедитесь, что бронь сохранена. Запрос создания без токена или с просроченным токеном вернёт 401.

Тесты. Скопируйте четыре файла в корень auth_owned_demo:

# conftest.py
import os
import pytest

os.environ.setdefault("SECRET_KEY", "local-test-settings-only")
from fastapi.testclient import TestClient
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.database import create_db_engine, get_db
from app.models import Base
from app.settings import Settings, get_settings

@pytest.fixture
def client(tmp_path):
    database_path = tmp_path / "test.db"
    test_engine = create_db_engine(f"sqlite:///{database_path.as_posix()}")
    previous_overrides = dict(app.dependency_overrides)
    try:
        Base.metadata.create_all(bind=test_engine)
        TestingSession = sessionmaker(bind=test_engine, autocommit=False, autoflush=False)

        def override_get_db():
            db = TestingSession()
            try:
                yield db
            finally:
                db.close()

        app.dependency_overrides[get_db] = override_get_db
        app.dependency_overrides[get_settings] = lambda: Settings(secret_key="test-only-key-not-for-public-deploy")
        with TestClient(app) as test_client:
            yield test_client
    finally:
        app.dependency_overrides.clear()
        app.dependency_overrides.update(previous_overrides)
        test_engine.dispose()
        database_path.unlink(missing_ok=True)

# auth_test_helpers.py
def login_headers(client, email="owner@example.com"):
    credentials = {"email": email, "password": "student-pass-123"}
    registered = client.post("/api/users", json=credentials)
    assert registered.status_code == 201
    response = client.post("/api/auth/login", json=credentials)
    assert response.status_code == 200
    return {"Authorization": "Bearer " + response.json()["access_token"]}

# test_bookings.py
import pytest
from auth_test_helpers import login_headers

@pytest.mark.parametrize("case,expected_status,expected_detail", [ ("missing_room", 404, "Room not found"), ("naive_start", 422, ["body", "starts_at"]), ("conflict", 409, "Booking conflict"), ]) def test_booking_errors(client, case, expected_status, expected_detail): headers = login_headers(client) room_id = 999 body = {"starts_at": "2025-10-01T10:00:00Z", "ends_at": "2025-10-01T11:00:00Z"} if case != "missing_room": created = client.post("/api/rooms", json={"name": "Альфа", "capacity": 6, "address": {"city": "Москва", "street": "Учебная, 1"}}) assert created.status_code == 201 room_id = created.json()["id"] path = f"/api/rooms/{room_id}/bookings" if case == "conflict": first = client.post(path, json=body, headers=headers) assert first.status_code == 201 if case == "naive_start": body["starts_at"] = "2025-10-01T10:00:00"

    response = client.post(path, json=body, headers=headers)
    assert response.status_code == expected_status
    detail = response.json()["detail"]
    if expected_status == 422:
        assert isinstance(detail, list)
        assert any(error.get("loc") == expected_detail for error in detail)
    else:
        assert detail == expected_detail
    if case == "conflict":
        listing = client.get(path)
        assert listing.status_code == 200
        assert listing.json() == [first.json()]

# test_auth.py
from datetime import datetime, timedelta, timezone
import jwt
import pytest
from auth_test_helpers import login_headers

@pytest.mark.parametrize("case", ["missing", "invalid", "expired", "no_exp", "bad_sub", "missing_user"]) def test_rejected_tokens(client, case): key = "test-only-key-not-for-public-deploy" payload = {"sub": "999999", "exp": datetime.now(timezone.utc) + timedelta(minutes=5)} headers = {} if case == "invalid": headers = {"Authorization": "Bearer invalid.token.here"} elif case != "missing": if case == "expired": payload["exp"] = datetime.now(timezone.utc) - timedelta(minutes=1) elif case == "no_exp": payload.pop("exp") elif case == "bad_sub": payload["sub"] = 17 headers = {"Authorization": "Bearer " + jwt.encode(payload, key, algorithm="HS256")} response = client.post("/api/rooms/999999/bookings", headers=headers, json={"starts_at": "2025-10-01T10:00:00Z", "ends_at": "2025-10-01T11:00:00Z"}) assert response.status_code == 401 assert response.headers["www-authenticate"] == "Bearer"

def test_valid_token_creates_booking(client):
    headers = login_headers(client)
    created = client.post("/api/rooms", json={"name": "Alpha", "capacity": 6,
        "address": {"city": "Moscow", "street": "Test, 1"}})
    assert created.status_code == 201
    room_id = created.json()["id"]
    response = client.post(f"/api/rooms/{room_id}/bookings", headers=headers,
        json={"starts_at": "2025-10-01T10:00:00Z", "ends_at": "2025-10-01T11:00:00Z"})
    assert response.status_code == 201
    assert response.json()["room_id"] == room_id

Фикстура client подменяет get_db и get_settings, но не get_current_user: зависимость для проверки токена работает в тестах так же, как в продакшене. Случай missing_user использует токен с sub, которого нет ни в одной строке таблицы users тестовой базы — это и есть проверка последней ветки get_current_user. Прежние overrides восстанавливаются в блоке finally, поэтому тесты не влияют друг на друга.

Во всех случаях проверки токена путь содержит целочисленный room_id, а тело запроса валидно. Так тесты изолируют ошибки credentials: оснований для 422 в этих запросах нет. Каждый перечисленный дефект токена должен дать 401.