Пересечение интервалов бронирования в FastAPI: задача

Проверка занятости переговорной

В Roomly нужно найти пересечение предложенного интервала с уже сохранёнными бронированиями одной комнаты. Учитывайте и частичное пересечение, и полное включение одного интервала в другой. Совпадение конца одной встречи с началом следующей конфликтом не считается.

Верните одно найденное конфликтующее бронирование или сообщите, что пересечений нет. Брони других комнат не должны влиять на результат. Функция только читает записи; она не проверяет существование самой комнаты и не подтверждает возможность бронирования целиком.

Что уже дано

Это самостоятельная задача на функцию, без HTTP-маршрутов и без изменения прежней базы. Перед вашим кодом уже исполняется следующий готовый фрагмент. DATABASE_URL указывает на новый собственный SQLite-файл в отдельном временном каталоге; создавать пути, схемы и Session самостоятельно не нужно.

from datetime import datetime, timezone
from sqlalchemy import create_engine, event, select, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker, Session

class Base(DeclarativeBase):
    pass

class Room(Base):
    __tablename__ = "rooms"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    capacity: Mapped[int]
    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"))
    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")

engine = create_engine(DATABASE_URL)

@event.listens_for(engine, "connect")
def enable_foreign_keys(connection, connection_record):
    connection.execute("PRAGMA foreign_keys=ON")

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def to_db_utc(value: datetime) -> datetime:
    if value.utcoffset() is None:
        raise ValueError("Timezone required")
    return value.astimezone(timezone.utc).replace(tzinfo=None)

Публичный переходник создаёт настоящую Session, разбирает переданные диагностические ISO-строки в aware datetime, вызывает вашу find_conflict и возвращает JSON-наблюдение результата. Он не вычисляет пересечение. Другой диагностический вызов читает тот же файл независимым SQLite-соединением; наблюдатель отмечает фактические SQL-записи. Переходник создаёт только данные, поступившие из скрытой проверки, а после работы закрывает ресурсы и удаляет свой временный каталог. Писать эти средства не нужно.

Что нужно сделать

Реализуйте find_conflict(db, room_id, new_start, new_end). db — открытая Session данной базы; границы — уже проверенные непустые aware datetime, но смещение может быть любым. Через готовый to_db_utc приведите обе границы к naive UTC и выполните SQLAlchemy-запрос к Booking: та же room_id, существующее начало строго раньше нового конца и существующий конец строго позже нового начала.

Верните один найденный ORM-объект Booking из переданной Session либо None. Если совпадений несколько, подходит любой из них, порядок не требуется. Соседние интервалы не пересекаются; полное вложение, частичное пересечение и одинаковые интервалы пересекаются. Отсутствующей комнате не соответствуют строки, поэтому для неё результат None: это не проверка существования Room и не HTTP404.

Функция только читает: не изменяйте ORM-объекты, не выполняйте INSERT/UPDATE/DELETE, не добавляйте и не удаляйте записи. Повторный вызов должен учитывать актуальные строки, в том числе добавленные между обращениями. Новую валидацию интервала и обработчик бронирования писать не нужно.

Ввод и вывод

stdin не используется. Печатать через print в stdout ничего не нужно. После вашего кода проверка вызывает верхнеуровневую функцию find_conflict через публичный RPC. Ваша функция получает обычные Python-объекты Session и datetime и возвращает Booking или None; JSON нужен только служебному переходнику. Не реализуйте app, HTTP, Uvicorn или чтение stdin.

Если у Room3 есть Booking19 с интервалом [10:00,12:00) UTC, новый [11:00,13:00) UTC возвращает этот объект, а [12:00,13:00) — None. Если в интервал попадают две её брони, любой из этих объектов допустим.