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

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.