Текущий пользователь как 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-схемы.