codex_services.booking.slot_master

slot_master

Booking engine — chain service scheduling with backtracking.

Classes

BookingEngineError

Bases: Exception

Base exception of the booking engine.

All other exceptions inherit from it. A Django view can catch BookingEngineError for unified handling of all engine errors.

Example

try: result = service.book(...) except BookingEngineError as e: messages.error(request, str(e)) return redirect("booking:wizard")

Source code in src/codex_services/booking/_shared/exceptions.py

class BookingEngineError(Exception):
    """
    Base exception of the booking engine.

    All other exceptions inherit from it.
    A Django view can catch BookingEngineError for unified handling
    of all engine errors.

    Example:
        try:
            result = service.book(...)
        except BookingEngineError as e:
            messages.error(request, str(e))
            return redirect("booking:wizard")
    """

    default_message: str = "Booking system error"

    def __init__(self, message: str | None = None) -> None:
        super().__init__(message or self.default_message)

ChainBuildError

Bases: BookingEngineError

The engine could not assemble a chain for all services in the request.

Differs from NoAvailabilityError: here the chain was PARTIALLY assembled, but not to the end (constraint violation, incompatible services, etc.).

Used when

• max_chain_duration_minutes is exceeded
• Services are incompatible (future: excludes/tags)
• group_size > available parallel slots

Attributes:

Name	Type	Description
failed_at_index		Index of the service (in service_requests) where assembly failed.
reason		Technical reason (for logs).

Example

raise ChainBuildError( failed_at_index=2, reason="max_chain_duration_minutes=180 exceeded at service 'Coloring'", )

Source code in src/codex_services/booking/_shared/exceptions.py

class ChainBuildError(BookingEngineError):
    """
    The engine could not assemble a chain for all services in the request.

    Differs from NoAvailabilityError: here the chain was PARTIALLY assembled,
    but not to the end (constraint violation, incompatible services, etc.).

    Used when:
        - max_chain_duration_minutes is exceeded
        - Services are incompatible (future: excludes/tags)
        - group_size > available parallel slots

    Attributes:
        failed_at_index: Index of the service (in service_requests) where assembly failed.
        reason: Technical reason (for logs).

    Example:
        raise ChainBuildError(
            failed_at_index=2,
            reason="max_chain_duration_minutes=180 exceeded at service 'Coloring'",
        )
    """

    default_message = "Could not find a schedule for all selected services."

    def __init__(
        self,
        failed_at_index: int | None = None,
        reason: str | None = None,
        message: str | None = None,
    ) -> None:
        self.failed_at_index = failed_at_index
        self.reason = reason

        if message:
            final_message = message
        elif reason:
            final_message = f"Could not assemble the chain: {reason}."
        else:
            final_message = self.default_message

        super().__init__(final_message)

InvalidBookingDateError

Bases: BookingEngineError

Booking date is invalid.

Example reasons

• Date in the past
• Date beyond max_advance_days
• Salon is closed on this day

Attributes:

Name	Type	Description
booking_date		Problematic date.
reason		Human-readable explanation.

Example

raise InvalidBookingDateError( booking_date=date(2020, 1, 1), reason="Date in the past", )

Source code in src/codex_services/booking/_shared/exceptions.py

class InvalidBookingDateError(BookingEngineError):
    """
    Booking date is invalid.

    Example reasons:
        - Date in the past
        - Date beyond max_advance_days
        - Salon is closed on this day

    Attributes:
        booking_date: Problematic date.
        reason: Human-readable explanation.

    Example:
        raise InvalidBookingDateError(
            booking_date=date(2020, 1, 1),
            reason="Date in the past",
        )
    """

    default_message = "The selected date is unavailable for booking."

    def __init__(
        self,
        booking_date: date | None = None,
        reason: str | None = None,
        message: str | None = None,
    ) -> None:
        self.booking_date = booking_date
        self.reason = reason

        if message:
            final_message = message
        elif booking_date and reason:
            date_str = booking_date.strftime("%d.%m.%Y")
            final_message = f"Date {date_str} is unavailable: {reason}."
        elif booking_date:
            date_str = booking_date.strftime("%d.%m.%Y")
            final_message = f"Date {date_str} is unavailable for booking."
        else:
            final_message = self.default_message

        super().__init__(final_message)

InvalidServiceDurationError

Bases: BookingEngineError

Incorrect service duration.

Raised if duration_minutes <= 0 or exceeds the explicit maximum.

Attributes:

Name	Type	Description
service_id		ID of the problematic service.
duration_minutes		The passed duration value.

Example

raise InvalidServiceDurationError(service_id="5", duration_minutes=0)

str(e) -> "Service 5: incorrect duration 0 min."

Source code in src/codex_services/booking/_shared/exceptions.py

class InvalidServiceDurationError(BookingEngineError):
    """
    Incorrect service duration.

    Raised if duration_minutes <= 0 or exceeds the explicit maximum.

    Attributes:
        service_id: ID of the problematic service.
        duration_minutes: The passed duration value.

    Example:
        raise InvalidServiceDurationError(service_id="5", duration_minutes=0)
        # str(e) -> "Service 5: incorrect duration 0 min."
    """

    default_message = "Incorrect service duration."

    def __init__(
        self,
        service_id: str | None = None,
        duration_minutes: int | None = None,
        message: str | None = None,
    ) -> None:
        self.service_id = service_id
        self.duration_minutes = duration_minutes

        if message:
            final_message = message
        elif service_id is not None and duration_minutes is not None:
            final_message = (
                f"Service {service_id}: incorrect duration {duration_minutes} min. Duration must be greater than 0."
            )
        else:
            final_message = self.default_message

        super().__init__(final_message)

NoAvailabilityError

Bases: BookingEngineError

The engine found no schedule options for the request.

Raised when ChainFinder.find() returns an empty EngineResult. Translated to a user-friendly message in the Django view.

Attributes:

Name	Type	Description
booking_date		Date searched for.
service_ids		List of service IDs from the request.

Example

raise NoAvailabilityError( booking_date=date(2024, 5, 10), service_ids=["5", "12"], )

str(e) -> "No free slots on 10.05.2024 for the selected services."

Source code in src/codex_services/booking/_shared/exceptions.py

class NoAvailabilityError(BookingEngineError):
    """
    The engine found no schedule options for the request.

    Raised when ChainFinder.find() returns an empty EngineResult.
    Translated to a user-friendly message in the Django view.

    Attributes:
        booking_date: Date searched for.
        service_ids: List of service IDs from the request.

    Example:
        raise NoAvailabilityError(
            booking_date=date(2024, 5, 10),
            service_ids=["5", "12"],
        )
        # str(e) -> "No free slots on 10.05.2024 for the selected services."
    """

    default_message = "Unfortunately, there are no available slots for these services on the selected date."

    def __init__(
        self,
        booking_date: date | None = None,
        service_ids: list[str] | None = None,
        message: str | None = None,
    ) -> None:
        self.booking_date = booking_date
        self.service_ids = service_ids or []

        if message:
            final_message = message
        elif booking_date:
            date_str = booking_date.strftime("%d.%m.%Y")
            final_message = f"No free slots on {date_str} for the selected services. Please try choosing another date."
        else:
            final_message = self.default_message

        super().__init__(final_message)

ResourceNotAvailableError

Bases: BookingEngineError

Specific resource is not available for booking.

Used in RESOURCE_LOCKED mode when the selected resource does not work on this day or their schedule is empty.

Attributes:

Name	Type	Description
resource_id		Resource's ID.
booking_date		Date attempted to book.

Example

raise ResourceNotAvailableError(resource_id="3", booking_date=date(2024,5,10))

str(e) -> "Resource unavailable on 10.05.2024."

Source code in src/codex_services/booking/_shared/exceptions.py

class ResourceNotAvailableError(BookingEngineError):
    """
    Specific resource is not available for booking.

    Used in RESOURCE_LOCKED mode when the selected resource
    does not work on this day or their schedule is empty.

    Attributes:
        resource_id: Resource's ID.
        booking_date: Date attempted to book.

    Example:
        raise ResourceNotAvailableError(resource_id="3", booking_date=date(2024,5,10))
        # str(e) -> "Resource unavailable on 10.05.2024."
    """

    default_message = "The selected resource is unavailable on this date."

    def __init__(
        self,
        resource_id: str | None = None,
        booking_date: date | None = None,
        message: str | None = None,
    ) -> None:
        self.resource_id = resource_id
        self.booking_date = booking_date

        if message:
            final_message = message
        elif booking_date:
            date_str = booking_date.strftime("%d.%m.%Y")
            final_message = f"Resource unavailable on {date_str}. Please try another date."
        else:
            final_message = self.default_message

        super().__init__(final_message)

SlotAlreadyBookedError

Bases: BookingEngineError

Slot was free during display but became booked by the time of confirmation.

Race condition: client A and client B are viewing the same slot simultaneously. A clicks "Book" first -> B receives this error.

Attributes:

Name	Type	Description
resource_id		Resource ID.
service_id		Service ID.
booking_date		Booking date.
slot_time		Selected time (string "HH:MM").

Example

raise SlotAlreadyBookedError( resource_id="3", service_id="5", booking_date=date(2024, 5, 10), slot_time="14:00", )

str(e) -> "Slot 14:00 on 10.05.2024 was booked. Please choose another time."

Source code in src/codex_services/booking/_shared/exceptions.py

class SlotAlreadyBookedError(BookingEngineError):
    """
    Slot was free during display but became booked by the time of confirmation.

    Race condition: client A and client B are viewing the same slot simultaneously.
    A clicks "Book" first -> B receives this error.

    Attributes:
        resource_id: Resource ID.
        service_id: Service ID.
        booking_date: Booking date.
        slot_time: Selected time (string "HH:MM").

    Example:
        raise SlotAlreadyBookedError(
            resource_id="3",
            service_id="5",
            booking_date=date(2024, 5, 10),
            slot_time="14:00",
        )
        # str(e) -> "Slot 14:00 on 10.05.2024 was booked. Please choose another time."
    """

    default_message = "The selected slot was booked. Please choose another time."

    def __init__(
        self,
        resource_id: str | None = None,
        service_id: str | None = None,
        booking_date: date | None = None,
        slot_time: str | None = None,
        message: str | None = None,
    ) -> None:
        self.resource_id = resource_id
        self.service_id = service_id
        self.booking_date = booking_date
        self.slot_time = slot_time

        if message:
            final_message = message
        elif slot_time and booking_date:
            date_str = booking_date.strftime("%d.%m.%Y")
            final_message = (
                f"Slot {slot_time} on {date_str} was booked while you were processing. Please choose another time."
            )
        else:
            final_message = self.default_message

        super().__init__(final_message)

AvailabilityProvider

Bases: Protocol

Full availability provider — the main adapter contract. Implement this protocol for each framework adapter (Django, SQLAlchemy, etc.).

Source code in src/codex_services/booking/_shared/interfaces.py

class AvailabilityProvider(Protocol):
    """
    Full availability provider — the main adapter contract.
    Implement this protocol for each framework adapter (Django, SQLAlchemy, etc.).
    """

    def build_resources_availability(
        self,
        resource_ids: list[str],
        target_date: date,
        cache_ttl: int = 300,
        exclude_appointment_ids: list[int] | None = None,
    ) -> dict[str, ResourceAvailability]:
        """Build availability for a single date. Return {resource_id: ResourceAvailability}."""
        ...

    def build_availability_batch(
        self,
        resource_ids: list[str],
        start_date: date,
        end_date: date,
    ) -> dict[date, dict[str, ResourceAvailability]]:
        """
        Build availability for a date range in a single batch.
        Must avoid N+1 queries — group appointments in memory.
        Return {date: {resource_id: ResourceAvailability}}.
        """
        ...

Functions

build_resources_availability(resource_ids, target_date, cache_ttl=300, exclude_appointment_ids=None)

Build availability for a single date. Return {resource_id: ResourceAvailability}.

Source code in src/codex_services/booking/_shared/interfaces.py

def build_resources_availability(
    self,
    resource_ids: list[str],
    target_date: date,
    cache_ttl: int = 300,
    exclude_appointment_ids: list[int] | None = None,
) -> dict[str, ResourceAvailability]:
    """Build availability for a single date. Return {resource_id: ResourceAvailability}."""
    ...

build_availability_batch(resource_ids, start_date, end_date)

Build availability for a date range in a single batch. Must avoid N+1 queries — group appointments in memory. Return {date: {resource_id: ResourceAvailability}}.

Source code in src/codex_services/booking/_shared/interfaces.py

def build_availability_batch(
    self,
    resource_ids: list[str],
    start_date: date,
    end_date: date,
) -> dict[date, dict[str, ResourceAvailability]]:
    """
    Build availability for a date range in a single batch.
    Must avoid N+1 queries — group appointments in memory.
    Return {date: {resource_id: ResourceAvailability}}.
    """
    ...

BusySlotsProvider

Bases: Protocol

Provides busy time slots for resources.

Source code in src/codex_services/booking/_shared/interfaces.py

class BusySlotsProvider(Protocol):
    """Provides busy time slots for resources."""

    def get_busy_intervals(
        self, resource_ids: list[str], target_date: date
    ) -> dict[str, list[tuple[datetime, datetime]]]:
        """Return {resource_id: [(start, end), ...]} of busy times."""
        ...

Functions

get_busy_intervals(resource_ids, target_date)

Return {resource_id: [(start, end), ...]} of busy times.

Source code in src/codex_services/booking/_shared/interfaces.py

def get_busy_intervals(
    self, resource_ids: list[str], target_date: date
) -> dict[str, list[tuple[datetime, datetime]]]:
    """Return {resource_id: [(start, end), ...]} of busy times."""
    ...

ScheduleProvider

Bases: Protocol

Provides working schedules for resources.

Source code in src/codex_services/booking/_shared/interfaces.py

class ScheduleProvider(Protocol):
    """Provides working schedules for resources."""

    def get_working_hours(self, resource_id: str, target_date: date) -> tuple[time, time] | None:
        """Return (start, end) of working day or None if day off."""
        ...

    def get_break_interval(self, resource_id: str, target_date: date) -> tuple[datetime, datetime] | None:
        """Return (start, end) of break or None."""
        ...

Functions

get_working_hours(resource_id, target_date)

Return (start, end) of working day or None if day off.

Source code in src/codex_services/booking/_shared/interfaces.py

def get_working_hours(self, resource_id: str, target_date: date) -> tuple[time, time] | None:
    """Return (start, end) of working day or None if day off."""
    ...

get_break_interval(resource_id, target_date)

Return (start, end) of break or None.

Source code in src/codex_services/booking/_shared/interfaces.py

def get_break_interval(self, resource_id: str, target_date: date) -> tuple[datetime, datetime] | None:
    """Return (start, end) of break or None."""
    ...

ChainFinder

Finds combinations of time slots for N services (booking chains).

Core algorithm: recursive backtracking. Iterates over possible resources and their free windows for each service. Checks for the absence of conflicts with already assigned services in the chain.

Examples:

1 service, any free resource:

finder = ChainFinder(step_minutes=30)
request = BookingEngineRequest(
    service_requests=[
        ServiceRequest(service_id="5", duration_minutes=60,
                       possible_resource_ids=["1", "2"])
    ],
    booking_date=date(2024, 5, 10),
    mode=BookingMode.SINGLE_DAY,
)
result = finder.find(request, resources_availability)
# result.solutions -- list of BookingChainSolution
# result.get_unique_start_times() -> ["09:00", "09:30", "10:00", ...]

2 services on the same day:

request = BookingEngineRequest(
    service_requests=[svc_task_1, svc_task_2],
    booking_date=date(2024, 5, 10),
    mode=BookingMode.SINGLE_DAY,
)
result = finder.find(request, availability)

Booking a specific resource (RESOURCE_LOCKED):

# Just pass possible_resource_ids=[locked_resource_id]
# and mode=BookingMode.RESOURCE_LOCKED

Parameters:

Name	Type	Description	Default
step_minutes	int	Grid slot step (30 min by default).	30
min_start	datetime | None	Minimum acceptable start time of the first service. None = no restriction (e.g., in tests).	None

Source code in src/codex_services/booking/slot_master/chain_finder.py

class ChainFinder:
    """
    Finds combinations of time slots for N services (booking chains).

    Core algorithm: recursive backtracking.
    Iterates over possible resources and their free windows for each service.
    Checks for the absence of conflicts with already assigned services in the chain.

    Examples:
        1 service, any free resource:
        ```python
        finder = ChainFinder(step_minutes=30)
        request = BookingEngineRequest(
            service_requests=[
                ServiceRequest(service_id="5", duration_minutes=60,
                               possible_resource_ids=["1", "2"])
            ],
            booking_date=date(2024, 5, 10),
            mode=BookingMode.SINGLE_DAY,
        )
        result = finder.find(request, resources_availability)
        # result.solutions -- list of BookingChainSolution
        # result.get_unique_start_times() -> ["09:00", "09:30", "10:00", ...]
        ```

        2 services on the same day:
        ```python
        request = BookingEngineRequest(
            service_requests=[svc_task_1, svc_task_2],
            booking_date=date(2024, 5, 10),
            mode=BookingMode.SINGLE_DAY,
        )
        result = finder.find(request, availability)
        ```

        Booking a specific resource (RESOURCE_LOCKED):
        ```python
        # Just pass possible_resource_ids=[locked_resource_id]
        # and mode=BookingMode.RESOURCE_LOCKED
        ```

    Args:
        step_minutes: Grid slot step (30 min by default).
        min_start: Minimum acceptable start time of the first service.
                   None = no restriction (e.g., in tests).
    """

    def __init__(
        self,
        step_minutes: int = 30,
        min_start: datetime | None = None,
    ) -> None:
        self.step_minutes = step_minutes
        self.min_start = min_start
        self._calc = SlotCalculator(step_minutes)

    def find(
        self,
        request: BookingEngineRequest,
        resources_availability: dict[str, MasterAvailability],
        max_solutions: int = 50,
        max_unique_starts: int | None = None,
    ) -> EngineResult:
        """
        Unified engine entry point. Delegates to the appropriate mode.

        Args:
            request: Input request with services, date, and mode.
            resources_availability: Dictionary {resource_id: MasterAvailability}.
                                    Usually prepared by an AvailabilityAdapter.
                                    Keys -- strings (resource identifiers).
            max_solutions: Maximum number of options the engine will return.
                           Does not affect correctness -- only completeness.
            max_unique_starts: Stop after finding N unique start times
                               (based on items[0].start_time).
                               None = no limit.
                               Example: max_unique_starts=8 -> only the closest 8 slots,
                               even if there are 16 in a day. Halves engine iterations.

        Returns:
            EngineResult with found solutions.
            solutions sorted by the start time of the first service.
        """
        # Fail Fast: Check for unsupported features
        if request.group_size > 1:
            raise NotImplementedError("Group bookings (group_size > 1) are not yet supported.")

        if request.mode == BookingMode.MULTI_DAY:
            raise NotImplementedError("MULTI_DAY booking mode is not yet implemented.")

        if request.mode in (BookingMode.SINGLE_DAY, BookingMode.RESOURCE_LOCKED):
            solutions = self._find_single_day(request, resources_availability, max_solutions, max_unique_starts)
        else:
            solutions = []  # pragma: no cover

        solutions.sort(key=lambda s: s.starts_at)
        return EngineResult(mode=request.mode, solutions=solutions)

    def find_nearest(
        self,
        request: BookingEngineRequest,
        get_availability_for_date: Callable[[date], dict[str, MasterAvailability]],
        search_from: date,
        search_days: int = 60,
        max_solutions_per_day: int = 1,
    ) -> EngineResult:
        """
        Searches for the first day with available slots in the search_days range.

        Used for:
            - Rebooking: resource is sick -> find a new date for N appointments
            - Waitlist: closest free slot to notify the client
            - MULTI_DAY planning: find the first day the chain fits

        Args:
            request: Request (booking_date will be replaced for each checked day).
            get_availability_for_date: callable(date) -> dict[str, MasterAvailability].
                                       Called for each checked day.
                                       Wraps DjangoAvailabilityAdapter in the Django layer.
            search_from: Date to start the search from (inclusive).
            search_days: Maximum days to check. Defaults to 60.
            max_solutions_per_day: How many solutions to look for per day.
                                   1 = fast mode (stop at the first one).

        Returns:
            EngineResult of the first day with solutions.
            If nothing found in search_days — EngineResult(solutions=[]).

        Example (Django layer):
            adapter = DjangoAvailabilityAdapter()
            resource_ids = [...]

            def get_avail(d):
                return adapter.build_resources_availability(resource_ids, d)

            result = finder.find_nearest(request, get_avail, search_from=date.today())
            if result.has_solutions:
                print(result.best.starts_at)  # date and time of the new slot
        """
        for offset in range(search_days):
            check_date = search_from + timedelta(days=offset)

            # Update date in the request (frozen=True -> model_copy)
            day_request = request.model_copy(update={"booking_date": check_date})

            availability = get_availability_for_date(check_date)
            if not availability:
                continue

            result = self.find(day_request, availability, max_solutions=max_solutions_per_day)
            if result.has_solutions:
                return result

        return EngineResult(mode=request.mode, solutions=[])

    # ---------------------------------------------------------------------------
    # SINGLE_DAY / RESOURCE_LOCKED mode
    # ---------------------------------------------------------------------------

    def _find_single_day(
        self,
        request: BookingEngineRequest,
        resources_availability: dict[str, MasterAvailability],
        max_solutions: int,
        max_unique_starts: int | None = None,
    ) -> list[BookingChainSolution]:
        """
        Search for a chain for the 'all services in one day' mode.

        Performance:
            Internally works with _SlotCandidate (__slots__) -- without Pydantic.
            Pydantic objects are created ONLY for final solutions.
            With 3 services x 3 resources x 20 slots -- up to 1800 iterations.

        Request parameters considered in this method:
            request.overlap_allowed:
                True  -> different resources work independently (can be parallel).
                False -> each subsequent service starts only after the previous one ends.
            request.max_chain_duration_minutes:
                If set — cuts off backtracking branches where the chain already exceeds the limit.
        """
        solutions: list[BookingChainSolution] = []
        chain: list[_SlotCandidate] = []
        seen_starts: set[str] = set()  # unique start times of the first service

        def backtrack(service_index: int) -> None:
            if len(solutions) >= max_solutions:
                return
            if max_unique_starts is not None and len(seen_starts) >= max_unique_starts:
                return

            if service_index >= len(request.service_requests):
                # Final check + conversion to Pydantic only here
                if self._no_conflicts_fast(chain):
                    solution = BookingChainSolution(items=[c.to_solution() for c in chain])
                    solutions.append(solution)
                    seen_starts.add(chain[0].start_time.strftime("%H:%M"))
                return

            service_req = request.service_requests[service_index]
            duration_delta = timedelta(minutes=service_req.duration_minutes)
            gap_delta = timedelta(minutes=service_req.min_gap_after_minutes)

            # --- Parallel Group Logic ---
            # If the current service has a parallel_group, look for a "partner" in the already assembled chain.
            # If found, the start time must strictly match.
            forced_start_time: datetime | None = None
            if service_req.parallel_group:
                for item in chain:
                    if item.parallel_group == service_req.parallel_group:
                        forced_start_time = item.start_time
                        break

            for resource_id in service_req.possible_resource_ids:
                availability = resources_availability.get(resource_id)
                if not availability:
                    continue

                # Busy intervals of this resource in the current chain
                resource_busy = [(c.start_time, c.gap_end_time) for c in chain if c.resource_id == resource_id]

                # If there is a forced_start_time, check only it
                if forced_start_time:
                    # Check if the resource is free at this specific time
                    slot_end = forced_start_time + duration_delta
                    gap_end = slot_end + gap_delta

                    # 1. Check resource's occupancy in the chain
                    if not self._is_slot_free_fast(forced_start_time, gap_end, resource_busy):
                        continue

                    # 2. Check if it falls into the resource's free windows
                    in_window = False
                    for w_start, w_end in availability.free_windows:
                        if w_start <= forced_start_time and w_end >= slot_end:
                            in_window = True
                            break

                    if not in_window:
                        continue

                    # If everything is ok - add and move on
                    chain.append(
                        _SlotCandidate(
                            service_id=service_req.service_id,
                            resource_id=resource_id,
                            start_time=forced_start_time,
                            end_time=slot_end,
                            gap_end_time=gap_end,
                            parallel_group=service_req.parallel_group,
                        )
                    )
                    backtrack(service_index + 1)
                    chain.pop()
                    continue  # Move to the next resource, no need to iterate over slots

                # --- Standard Logic (No forced start time) ---
                effective_min = self._effective_min_start(resource_busy, availability.buffer_between_minutes)

                # overlap_allowed=False: each service starts after the end of all previous ones
                if not request.overlap_allowed and chain:
                    chain_ends_at = max(c.end_time for c in chain)
                    if effective_min is None or effective_min < chain_ends_at:
                        effective_min = chain_ends_at

                # --- Anchor Logic ---
                # Use anchor only for the first service or if services are parallel.
                # For 2nd+ service in a sequential chain, anchor is None (tight fit).
                use_anchor = availability.work_start if (not chain or request.overlap_allowed) else None

                for window_start, window_end in availability.free_windows:
                    slots = self._calc.find_slots_in_window(
                        window_start=window_start,
                        window_end=window_end,
                        duration_minutes=service_req.duration_minutes,
                        min_start=effective_min,
                        grid_anchor=use_anchor,
                    )

                    for slot_start in slots:
                        if len(solutions) >= max_solutions:
                            return
                        if max_unique_starts is not None and len(seen_starts) >= max_unique_starts:
                            return

                        slot_end = slot_start + duration_delta
                        gap_end = slot_end + gap_delta

                        # Simple datetime comparison -- without Pydantic
                        if not self._is_slot_free_fast(slot_start, gap_end, resource_busy):  # pragma: no cover
                            continue  # pragma: no cover

                        # Check maximum chain duration
                        if request.max_chain_duration_minutes is not None and chain:
                            chain_start = min(c.start_time for c in chain)
                            prospective_span = int(
                                (max(slot_end, max(c.end_time for c in chain)) - chain_start).total_seconds() / 60
                            )
                            if prospective_span > request.max_chain_duration_minutes:
                                continue  # cut off branch - chain is too long

                        chain.append(
                            _SlotCandidate(
                                service_id=service_req.service_id,
                                resource_id=resource_id,
                                start_time=slot_start,
                                end_time=slot_end,
                                gap_end_time=gap_end,
                                parallel_group=service_req.parallel_group,
                            )
                        )
                        backtrack(service_index + 1)
                        chain.pop()

        backtrack(0)
        return solutions

    # ---------------------------------------------------------------------------
    # Fast internal checks (without Pydantic -- stdlib only)
    # ---------------------------------------------------------------------------

    @staticmethod
    def _is_slot_free_fast(
        slot_start: datetime,
        slot_end: datetime,
        busy_intervals: list[tuple[datetime, datetime]],
    ) -> bool:
        """Fast slot availability check. Without Pydantic."""
        return all(not (slot_start < b_end and slot_end > b_start) for b_start, b_end in busy_intervals)

    @staticmethod
    def _no_conflicts_fast(chain: list["_SlotCandidate"]) -> bool:
        """Final check of the chain for resource conflicts. Without Pydantic."""
        by_resource: dict[str, list[_SlotCandidate]] = {}
        for c in chain:
            if c.resource_id not in by_resource:
                by_resource[c.resource_id] = []
            by_resource[c.resource_id].append(c)

        for slots in by_resource.values():
            if len(slots) < 2:
                continue
            sorted_slots = sorted(slots, key=lambda s: s.start_time)
            for i in range(len(sorted_slots) - 1):
                if sorted_slots[i + 1].start_time < sorted_slots[i].gap_end_time:  # pragma: no cover
                    return False  # pragma: no cover

        return True

    def _effective_min_start(
        self,
        resource_busy: list[tuple[datetime, datetime]],
        buffer_minutes: int,
    ) -> datetime | None:
        """
        Minimum allowable start time for the resource.

        Considers:
            - self.min_start (global -- "no earlier than N mins from now")
            - End of the last booked slot + buffer between clients
        """
        candidates: list[datetime] = []

        if self.min_start:
            candidates.append(self.min_start)

        if resource_busy:
            last_end = max(end for _, end in resource_busy)
            candidates.append(last_end + timedelta(minutes=buffer_minutes))

        return max(candidates) if candidates else None

Functions

find(request, resources_availability, max_solutions=50, max_unique_starts=None)

Unified engine entry point. Delegates to the appropriate mode.

Parameters:

Name	Type	Description	Default
request	BookingEngineRequest	Input request with services, date, and mode.	required
resources_availability	dict[str, MasterAvailability]	Dictionary {resource_id: MasterAvailability}. Usually prepared by an AvailabilityAdapter. Keys -- strings (resource identifiers).	required
max_solutions	int	Maximum number of options the engine will return. Does not affect correctness -- only completeness.	50
max_unique_starts	int | None	Stop after finding N unique start times (based on items[0].start_time). None = no limit. Example: max_unique_starts=8 -> only the closest 8 slots, even if there are 16 in a day. Halves engine iterations.	None

Returns:

Type	Description
EngineResult	EngineResult with found solutions.
EngineResult	solutions sorted by the start time of the first service.

Source code in src/codex_services/booking/slot_master/chain_finder.py

def find(
    self,
    request: BookingEngineRequest,
    resources_availability: dict[str, MasterAvailability],
    max_solutions: int = 50,
    max_unique_starts: int | None = None,
) -> EngineResult:
    """
    Unified engine entry point. Delegates to the appropriate mode.

    Args:
        request: Input request with services, date, and mode.
        resources_availability: Dictionary {resource_id: MasterAvailability}.
                                Usually prepared by an AvailabilityAdapter.
                                Keys -- strings (resource identifiers).
        max_solutions: Maximum number of options the engine will return.
                       Does not affect correctness -- only completeness.
        max_unique_starts: Stop after finding N unique start times
                           (based on items[0].start_time).
                           None = no limit.
                           Example: max_unique_starts=8 -> only the closest 8 slots,
                           even if there are 16 in a day. Halves engine iterations.

    Returns:
        EngineResult with found solutions.
        solutions sorted by the start time of the first service.
    """
    # Fail Fast: Check for unsupported features
    if request.group_size > 1:
        raise NotImplementedError("Group bookings (group_size > 1) are not yet supported.")

    if request.mode == BookingMode.MULTI_DAY:
        raise NotImplementedError("MULTI_DAY booking mode is not yet implemented.")

    if request.mode in (BookingMode.SINGLE_DAY, BookingMode.RESOURCE_LOCKED):
        solutions = self._find_single_day(request, resources_availability, max_solutions, max_unique_starts)
    else:
        solutions = []  # pragma: no cover

    solutions.sort(key=lambda s: s.starts_at)
    return EngineResult(mode=request.mode, solutions=solutions)

find_nearest(request, get_availability_for_date, search_from, search_days=60, max_solutions_per_day=1)

Searches for the first day with available slots in the search_days range.

Used for

• Rebooking: resource is sick -> find a new date for N appointments
• Waitlist: closest free slot to notify the client
• MULTI_DAY planning: find the first day the chain fits

Parameters:

Name	Type	Description	Default
request	BookingEngineRequest	Request (booking_date will be replaced for each checked day).	required
get_availability_for_date	Callable[[date], dict[str, MasterAvailability]]	callable(date) -> dict[str, MasterAvailability]. Called for each checked day. Wraps DjangoAvailabilityAdapter in the Django layer.	required
search_from	date	Date to start the search from (inclusive).	required
search_days	int	Maximum days to check. Defaults to 60.	60
max_solutions_per_day	int	How many solutions to look for per day. 1 = fast mode (stop at the first one).	1

Returns:

Type	Description
EngineResult	EngineResult of the first day with solutions.
EngineResult	If nothing found in search_days — EngineResult(solutions=[]).

Example (Django layer): adapter = DjangoAvailabilityAdapter() resource_ids = [...]

def get_avail(d):
    return adapter.build_resources_availability(resource_ids, d)

result = finder.find_nearest(request, get_avail, search_from=date.today())
if result.has_solutions:
    print(result.best.starts_at)  # date and time of the new slot

Source code in src/codex_services/booking/slot_master/chain_finder.py

def find_nearest(
    self,
    request: BookingEngineRequest,
    get_availability_for_date: Callable[[date], dict[str, MasterAvailability]],
    search_from: date,
    search_days: int = 60,
    max_solutions_per_day: int = 1,
) -> EngineResult:
    """
    Searches for the first day with available slots in the search_days range.

    Used for:
        - Rebooking: resource is sick -> find a new date for N appointments
        - Waitlist: closest free slot to notify the client
        - MULTI_DAY planning: find the first day the chain fits

    Args:
        request: Request (booking_date will be replaced for each checked day).
        get_availability_for_date: callable(date) -> dict[str, MasterAvailability].
                                   Called for each checked day.
                                   Wraps DjangoAvailabilityAdapter in the Django layer.
        search_from: Date to start the search from (inclusive).
        search_days: Maximum days to check. Defaults to 60.
        max_solutions_per_day: How many solutions to look for per day.
                               1 = fast mode (stop at the first one).

    Returns:
        EngineResult of the first day with solutions.
        If nothing found in search_days — EngineResult(solutions=[]).

    Example (Django layer):
        adapter = DjangoAvailabilityAdapter()
        resource_ids = [...]

        def get_avail(d):
            return adapter.build_resources_availability(resource_ids, d)

        result = finder.find_nearest(request, get_avail, search_from=date.today())
        if result.has_solutions:
            print(result.best.starts_at)  # date and time of the new slot
    """
    for offset in range(search_days):
        check_date = search_from + timedelta(days=offset)

        # Update date in the request (frozen=True -> model_copy)
        day_request = request.model_copy(update={"booking_date": check_date})

        availability = get_availability_for_date(check_date)
        if not availability:
            continue

        result = self.find(day_request, availability, max_solutions=max_solutions_per_day)
        if result.has_solutions:
            return result

    return EngineResult(mode=request.mode, solutions=[])

BookingChainSolution

Bases: BaseDTO

One complete solution for the entire request (set of slots for all services).

Found by the engine. Guarantees no conflicts between services and respects all resource availability constraints.

Fields

items (list[SingleServiceSolution]): List of slots in the order of service execution.

score (float): Quality score of the solution (higher is better). Can be influenced by preferred resources, idle time, or resource reuse.

Example

solution = BookingChainSolution(items=[slot1, slot2], score=10.0)
print(f"Booking span: {solution.span_minutes} minutes")

Source code in src/codex_services/booking/slot_master/dto.py

class BookingChainSolution(BaseDTO):
    """
    One complete solution for the entire request (set of slots for all services).

    Found by the engine. Guarantees no conflicts between services and
    respects all resource availability constraints.

    Fields:
        items (list[SingleServiceSolution]):
            List of slots in the order of service execution.

        score (float):
            Quality score of the solution (higher is better).
            Can be influenced by preferred resources, idle time, or resource reuse.

    Example:
        ```python
        solution = BookingChainSolution(items=[slot1, slot2], score=10.0)
        print(f"Booking span: {solution.span_minutes} minutes")
        ```
    """

    items: list[SingleServiceSolution] = Field(min_length=1)
    score: float = Field(default=0.0, description="Оценка качества решения")

    @property
    def starts_at(self) -> datetime:
        """Return the start time of the first service in the chain."""
        return min(s.start_time for s in self.items)

    @property
    def ends_at(self) -> datetime:
        """Return the end time of the last service (excluding gap)."""
        return max(s.end_time for s in self.items)

    @property
    def span_minutes(self) -> int:
        """Return total time from the start of the first to the end of the last service."""
        return int((self.ends_at - self.starts_at).total_seconds() / 60)

    def to_display(self) -> dict[str, Any]:
        """
        Convert the solution into a dictionary for UI/serialization.

        Returns:
            Dict: {service_id: {resource_id, start, end}, ...}
        """
        return {
            item.service_id: {
                "resource_id": item.resource_id,
                "start": item.start_time.strftime("%H:%M"),
                "end": item.end_time.strftime("%H:%M"),
            }
            for item in self.items
        }

    def __repr__(self) -> str:
        # GDPR Safe: Only structural info.
        return f"<BookingChainSolution score={self.score:.2f} items={self.items}>"

Attributes

starts_at property

Return the start time of the first service in the chain.

ends_at property

Return the end time of the last service (excluding gap).

span_minutes property

Return total time from the start of the first to the end of the last service.

Functions

to_display()

Convert the solution into a dictionary for UI/serialization.

Returns:

Name	Type	Description
Dict	dict[str, Any]	{service_id: {resource_id, start, end}, ...}

Source code in src/codex_services/booking/slot_master/dto.py

def to_display(self) -> dict[str, Any]:
    """
    Convert the solution into a dictionary for UI/serialization.

    Returns:
        Dict: {service_id: {resource_id, start, end}, ...}
    """
    return {
        item.service_id: {
            "resource_id": item.resource_id,
            "start": item.start_time.strftime("%H:%M"),
            "end": item.end_time.strftime("%H:%M"),
        }
        for item in self.items
    }

BookingEngineRequest

Bases: BookingRequest

Input request describing the entire desired booking chain.

Orchestrates multiple ServiceRequests into a single search task. Inherits booking_date from BookingRequest.

Fields

service_requests (list[ServiceRequest]): List of services to book. Order is critical for SINGLE_DAY mode — the engine will schedule them in the specified sequence. Minimum 1 service required.

booking_date (date): Target date. Inherited from BookingRequest. Used for SINGLE_DAY and RESOURCE_LOCKED. In MULTI_DAY mode, this represents the date of the first service.

mode (BookingMode): Engine operating strategy. Default is SINGLE_DAY.

overlap_allowed (bool): Allow parallel execution of services by different resources. False (default) — each subsequent service starts only after the previous one (plus its gap) ends. True — resources can work independently; services may start simultaneously if resources are available.

group_size (int): DEPRECATED. Use duplication of ServiceRequest with parallel_group.

max_chain_duration_minutes (int | None): Maximum total duration of the entire booking (from start of first to end of last service). None = no limit.

days_gap (list[int] | None): Day offsets for each service. Used strictly in MULTI_DAY mode.

Example

BookingEngineRequest(
    service_requests=[svc_1, svc_2],
    booking_date=date(2024, 5, 10),
    mode=BookingMode.SINGLE_DAY,
    overlap_allowed=True
)

Source code in src/codex_services/booking/slot_master/dto.py

class BookingEngineRequest(BookingRequest):
    """
    Input request describing the entire desired booking chain.

    Orchestrates multiple ServiceRequests into a single search task.
    Inherits booking_date from BookingRequest.

    Fields:
        service_requests (list[ServiceRequest]):
            List of services to book. Order is critical for SINGLE_DAY mode —
            the engine will schedule them in the specified sequence.
            Minimum 1 service required.

        booking_date (date):
            Target date. Inherited from BookingRequest.
            Used for SINGLE_DAY and RESOURCE_LOCKED.
            In MULTI_DAY mode, this represents the date of the first service.

        mode (BookingMode):
            Engine operating strategy. Default is SINGLE_DAY.

        overlap_allowed (bool):
            Allow parallel execution of services by different resources.
            False (default) — each subsequent service starts only after the
            previous one (plus its gap) ends.
            True — resources can work independently; services may start
            simultaneously if resources are available.

        group_size (int):
            DEPRECATED. Use duplication of ServiceRequest with parallel_group.

        max_chain_duration_minutes (int | None):
            Maximum total duration of the entire booking (from start of first
            to end of last service). None = no limit.

        days_gap (list[int] | None):
            Day offsets for each service. Used strictly in MULTI_DAY mode.

    Example:
        ```python
        BookingEngineRequest(
            service_requests=[svc_1, svc_2],
            booking_date=date(2024, 5, 10),
            mode=BookingMode.SINGLE_DAY,
            overlap_allowed=True
        )
        ```
    """

    service_requests: list[ServiceRequest] = Field(min_length=1, description="Минимум одна услуга")
    mode: BookingMode = BookingMode.SINGLE_DAY
    overlap_allowed: bool = Field(
        default=False,
        description="Разрешить параллельное выполнение услуг разными ресурсами",
    )
    group_size: int = Field(
        default=1,
        ge=1,
        description="DEPRECATED. Используйте дублирование ServiceRequest с parallel_group.",
    )
    max_chain_duration_minutes: int | None = Field(
        default=None,
        ge=1,
        description="Макс. длительность всей цепочки в минутах (None = без лимита)",
    )
    days_gap: list[int] | None = Field(
        default=None,
        description="Смещение в днях для каждой услуги (только MULTI_DAY)",
    )

    @property
    def total_duration_minutes(self) -> int:
        """Calculate total duration of all services without gap pauses."""
        return sum(s.duration_minutes for s in self.service_requests)

    @property
    def total_block_minutes(self) -> int:
        """
        Calculate total blocking time including pauses between services.
        Used for quick checks: does the chain fit into a given window.
        """
        return sum(s.total_block_minutes for s in self.service_requests)

    def __repr__(self) -> str:
        return f"<BookingEngineRequest date={self.booking_date} services={len(self.service_requests)}>"

Attributes

total_duration_minutes property

Calculate total duration of all services without gap pauses.

total_block_minutes property

Calculate total blocking time including pauses between services. Used for quick checks: does the chain fit into a given window.

EngineResult

Bases: BaseDTO

Engine work result containing all discovered solutions.

Fields

mode (BookingMode): The search strategy used. solutions (list[BookingChainSolution]): Found valid schedule options.

Example

result = ChainFinder().find(request, availability)
if result.has_solutions:
    print(f"Best start: {result.best.starts_at}")

Source code in src/codex_services/booking/slot_master/dto.py

class EngineResult(BaseDTO):
    """
    Engine work result containing all discovered solutions.

    Fields:
        mode (BookingMode): The search strategy used.
        solutions (list[BookingChainSolution]): Found valid schedule options.

    Example:
        ```python
        result = ChainFinder().find(request, availability)
        if result.has_solutions:
            print(f"Best start: {result.best.starts_at}")
        ```
    """

    mode: BookingMode
    solutions: list[BookingChainSolution] = Field(default_factory=list)

    @property
    def has_solutions(self) -> bool:
        """Return True if at least one solution was found."""
        return len(self.solutions) > 0

    @property
    def best(self) -> BookingChainSolution | None:
        """
        Return the primary solution.
        - Without scorer: earliest by start time.
        - After scoring: the option with the highest score.
        """
        return self.solutions[0] if self.solutions else None

    @property
    def best_scored(self) -> BookingChainSolution | None:
        """
        Return the option with the maximum score among all solutions.
        Differs from 'best' if the solution list hasn't been sorted yet.
        """
        if not self.solutions:
            return None
        return max(self.solutions, key=lambda s: s.score)

    def get_unique_start_times(self) -> list[str]:
        """
        Return unique start times of the first service for UI grid display.

        Returns:
            List[str]: ["09:00", "09:30", ...]
        """
        times = {s.starts_at.strftime("%H:%M") for s in self.solutions}
        return sorted(times)

    def __repr__(self) -> str:
        return f"<EngineResult mode={self.mode} solutions_count={len(self.solutions)}>"

Attributes

has_solutions property

Return True if at least one solution was found.

best property

Return the primary solution. - Without scorer: earliest by start time. - After scoring: the option with the highest score.

best_scored property

Return the option with the maximum score among all solutions. Differs from 'best' if the solution list hasn't been sorted yet.

Functions

get_unique_start_times()

Return unique start times of the first service for UI grid display.

Returns:

Type	Description
list[str]	List[str]: ["09:00", "09:30", ...]

Source code in src/codex_services/booking/slot_master/dto.py

def get_unique_start_times(self) -> list[str]:
    """
    Return unique start times of the first service for UI grid display.

    Returns:
        List[str]: ["09:00", "09:30", ...]
    """
    times = {s.starts_at.strftime("%H:%M") for s in self.solutions}
    return sorted(times)

MasterAvailability

Bases: ResourceAvailability

Available time windows of a resource for the slot-master booking type.

Inherits resource_id and free_windows from ResourceAvailability. Adds slot-master-specific fields: buffer_between_minutes and work_start.

Fields

resource_id (str): Resource identifier (inherited). free_windows (list[tuple[datetime, datetime]]): List of (start, end) tuples (inherited). buffer_between_minutes (int): Minimum buffer required between bookings. work_start (datetime | None): Shift start anchor for slot alignment.

Example

MasterAvailability(
    resource_id="resource_1",
    free_windows=[(datetime(2024,5,10,9,0), datetime(2024,5,10,12,0))],
    buffer_between_minutes=10,
)

Source code in src/codex_services/booking/slot_master/dto.py

class MasterAvailability(ResourceAvailability):
    """
    Available time windows of a resource for the slot-master booking type.

    Inherits resource_id and free_windows from ResourceAvailability.
    Adds slot-master-specific fields: buffer_between_minutes and work_start.

    Fields:
        resource_id (str): Resource identifier (inherited).
        free_windows (list[tuple[datetime, datetime]]): List of (start, end) tuples (inherited).
        buffer_between_minutes (int): Minimum buffer required between bookings.
        work_start (datetime | None): Shift start anchor for slot alignment.

    Example:
        ```python
        MasterAvailability(
            resource_id="resource_1",
            free_windows=[(datetime(2024,5,10,9,0), datetime(2024,5,10,12,0))],
            buffer_between_minutes=10,
        )
        ```
    """

    buffer_between_minutes: int = Field(default=0, ge=0)
    work_start: datetime | None = Field(
        default=None,
        description="Якорь для выравнивания сетки слотов (например, начало смены)",
    )

    def __repr__(self) -> str:
        return f"<MasterAvailability resource={self.resource_id} windows={len(self.free_windows)}>"

ServiceRequest

Bases: BaseDTO

Request for a single service within a booking chain.

The engine treats this as an atomic requirement. It operates with abstract resource IDs (possible_resource_ids) rather than specific business entities.

Fields

service_id (str): Unique identifier of the service. String for universality (can be "5", "uuid-xxx", "task-1" — doesn't matter).

duration_minutes (int): Duration of the service in minutes. Must be > 0.

min_gap_after_minutes (int): Minimum gap (minutes) after this service before the next one in the chain. Default is 0 (no gap). Example: if there is a cooling down period of 30 mins → min_gap_after_minutes=30.

possible_resource_ids (list[str]): List of resource IDs capable of performing this service. The engine chooses an available resource from this list. For RESOURCE_LOCKED mode, this should contain exactly one element.

parallel_group (str | None): Tag for parallel execution group. Services with the same parallel_group can be performed simultaneously by different resources (if overlap_allowed=True is set in the request). None = service is performed independently (standard sequential behavior).

Example

ServiceRequest(
    service_id="5",
    duration_minutes=60,
    possible_resource_ids=["1", "3", "7"],
)

Source code in src/codex_services/booking/slot_master/dto.py

class ServiceRequest(BaseDTO):
    """
    Request for a single service within a booking chain.

    The engine treats this as an atomic requirement. It operates with abstract
    resource IDs (possible_resource_ids) rather than specific business entities.

    Fields:
        service_id (str):
            Unique identifier of the service. String for universality
            (can be "5", "uuid-xxx", "task-1" — doesn't matter).

        duration_minutes (int):
            Duration of the service in minutes. Must be > 0.

        min_gap_after_minutes (int):
            Minimum gap (minutes) after this service before the next one
            in the chain. Default is 0 (no gap).
            Example: if there is a cooling down period of 30 mins → min_gap_after_minutes=30.

        possible_resource_ids (list[str]):
            List of resource IDs capable of performing this service.
            The engine chooses an available resource from this list.
            For RESOURCE_LOCKED mode, this should contain exactly one element.

        parallel_group (str | None):
            Tag for parallel execution group.
            Services with the same parallel_group can be performed simultaneously
            by different resources (if overlap_allowed=True is set in the request).
            None = service is performed independently (standard sequential behavior).

    Example:
        ```python
        ServiceRequest(
            service_id="5",
            duration_minutes=60,
            possible_resource_ids=["1", "3", "7"],
        )
        ```
    """

    service_id: str
    duration_minutes: int = Field(gt=0, description="Длительность в минутах")
    min_gap_after_minutes: int = Field(default=0, ge=0, description="Пауза после услуги перед следующей")
    possible_resource_ids: list[str] = Field(min_length=1, description="Хотя бы один ресурс должен быть указан")
    parallel_group: str | None = Field(
        default=None,
        description="Метка группы параллельного выполнения (одинаковый тег = одновременно)",
    )

    @property
    def total_block_minutes(self) -> int:
        """Return total time that blocks the resource: duration + gap after."""
        return self.duration_minutes + self.min_gap_after_minutes

    def __repr__(self) -> str:
        return f"<ServiceRequest id={self.service_id} dur={self.duration_minutes}>"

Attributes

total_block_minutes property

Return total time that blocks the resource: duration + gap after.

SingleServiceSolution

Bases: BookingSolution

Found slot for a single service in a booking chain.

Inherits resource_id, start_time, end_time from BookingSolution. Adds service_id and gap_end_time.

Fields

service_id (str): Reference to the original ServiceRequest.service_id. resource_id (str): Identifier of the resource assigned to this service (inherited). start_time (datetime): Scheduled start time of the service (inherited). end_time (datetime): Scheduled completion time (excluding gap) (inherited). gap_end_time (datetime): End of the blocking period (end_time + gap).

Example

slot = SingleServiceSolution(
    service_id="5",
    resource_id="1",
    start_time=datetime(2024, 5, 10, 10, 0),
    end_time=datetime(2024, 5, 10, 11, 0),
    gap_end_time=datetime(2024, 5, 10, 11, 15),
)

Source code in src/codex_services/booking/slot_master/dto.py

class SingleServiceSolution(BookingSolution):
    """
    Found slot for a single service in a booking chain.

    Inherits resource_id, start_time, end_time from BookingSolution.
    Adds service_id and gap_end_time.

    Fields:
        service_id (str): Reference to the original ServiceRequest.service_id.
        resource_id (str): Identifier of the resource assigned to this service (inherited).
        start_time (datetime): Scheduled start time of the service (inherited).
        end_time (datetime): Scheduled completion time (excluding gap) (inherited).
        gap_end_time (datetime): End of the blocking period (end_time + gap).

    Example:
        ```python
        slot = SingleServiceSolution(
            service_id="5",
            resource_id="1",
            start_time=datetime(2024, 5, 10, 10, 0),
            end_time=datetime(2024, 5, 10, 11, 0),
            gap_end_time=datetime(2024, 5, 10, 11, 15),
        )
        ```
    """

    service_id: str
    gap_end_time: datetime  # end_time + min_gap_after_minutes

    def __repr__(self) -> str:
        # GDPR Safe: Only IDs and Times. No notes, names, or PII.
        return (
            f"<SingleServiceSolution svc={self.service_id} "
            f"res={self.resource_id} "
            f"start={self.start_time.strftime('%H:%M')}>"
        )

WaitlistEntry

Bases: BaseDTO

Notification data for a nearest available slot for waitlisted clients.

Used when a desired slot was unavailable, but an alternative was found (e.g., via find_nearest() or a background worker).

Fields

available_date (date): Date of the found alternative slot. available_time (str): Start time of the first service ("HH:MM"). solution (BookingChainSolution): Complete schedule details. days_from_request (int): Delta from original request date for ranking.

Example

Worker detects a cancellation:

result = finder.find_nearest(request, search_from=original_date) if result.has_solutions: entry = WaitlistEntry.from_engine_result(result, original_date) notify_client(entry)

Source code in src/codex_services/booking/slot_master/dto.py

class WaitlistEntry(BaseDTO):
    """
    Notification data for a nearest available slot for waitlisted clients.

    Used when a desired slot was unavailable, but an alternative was found
    (e.g., via find_nearest() or a background worker).

    Fields:
        available_date (date): Date of the found alternative slot.
        available_time (str): Start time of the first service ("HH:MM").
        solution (BookingChainSolution): Complete schedule details.
        days_from_request (int): Delta from original request date for ranking.

    Example:
        # Worker detects a cancellation:
        result = finder.find_nearest(request, search_from=original_date)
        if result.has_solutions:
            entry = WaitlistEntry.from_engine_result(result, original_date)
            notify_client(entry)
    """

    available_date: date
    available_time: str  # "HH:MM"
    solution: BookingChainSolution
    days_from_request: int = 0

    @classmethod
    def from_engine_result(
        cls,
        result: "EngineResult",
        original_date: date,
    ) -> "WaitlistEntry | None":
        """
        Factory method to create a WaitlistEntry from an EngineResult.

        Args:
            result: The engine's output.
            original_date: Original requested date for delta calculation.

        Returns:
            WaitlistEntry or None if no solutions exist.
        """
        if not result.has_solutions:
            return None

        solution = result.best
        if solution is None:  # pragma: no cover
            return None  # pragma: no cover

        available_date = solution.starts_at.date()
        available_time = solution.starts_at.strftime("%H:%M")
        days_delta = (available_date - original_date).days

        return cls(
            available_date=available_date,
            available_time=available_time,
            solution=solution,
            days_from_request=max(0, days_delta),
        )

    def __repr__(self) -> str:
        return f"<WaitlistEntry date={self.available_date} time={self.available_time}>"

Functions

from_engine_result(result, original_date) classmethod

Factory method to create a WaitlistEntry from an EngineResult.

Parameters:

Name	Type	Description	Default
result	EngineResult	The engine's output.	required
original_date	date	Original requested date for delta calculation.	required

Returns:

Type	Description
WaitlistEntry | None	WaitlistEntry or None if no solutions exist.

Source code in src/codex_services/booking/slot_master/dto.py

@classmethod
def from_engine_result(
    cls,
    result: "EngineResult",
    original_date: date,
) -> "WaitlistEntry | None":
    """
    Factory method to create a WaitlistEntry from an EngineResult.

    Args:
        result: The engine's output.
        original_date: Original requested date for delta calculation.

    Returns:
        WaitlistEntry or None if no solutions exist.
    """
    if not result.has_solutions:
        return None

    solution = result.best
    if solution is None:  # pragma: no cover
        return None  # pragma: no cover

    available_date = solution.starts_at.date()
    available_time = solution.starts_at.strftime("%H:%M")
    days_delta = (available_date - original_date).days

    return cls(
        available_date=available_date,
        available_time=available_time,
        solution=solution,
        days_from_request=max(0, days_delta),
    )

BookingMode

Bases: StrEnum

Operating mode for ChainFinder.

SINGLE_DAY

All services from the request must fit within a single day. The common mode — multiple services in one visit.

MULTI_DAY

Each service can be scheduled for a different day (Stub).

RESOURCE_LOCKED

Booking for a specific resource (e.g., from their personal page). The engine relies entirely on the provided resource constraints.

Example

mode = BookingMode.SINGLE_DAY

Source code in src/codex_services/booking/slot_master/modes.py

class BookingMode(StrEnum):
    """
    Operating mode for ChainFinder.

    SINGLE_DAY:
        All services from the request must fit within a single day.
        The common mode — multiple services in one visit.

    MULTI_DAY:
        Each service can be scheduled for a different day (Stub).

    RESOURCE_LOCKED:
        Booking for a specific resource (e.g., from their personal page).
        The engine relies entirely on the provided resource constraints.

    Example:
        ```python
        mode = BookingMode.SINGLE_DAY
        ```
    """

    SINGLE_DAY = "single_day"
    MULTI_DAY = "multi_day"
    RESOURCE_LOCKED = "resource_locked"

BookingScorer

Evaluates engine solutions and returns an EngineResult with populated scores.

Usage

scorer = BookingScorer( weights=ScoringWeights(preferred_resource_bonus=15.0), preferred_resource_ids=["3", "7"], ) ranked = scorer.score(result)

Best solution by score (not just the earliest):

print(ranked.best.score) print(ranked.best.to_display())

All solutions sorted by score (highest -> first):

for solution in ranked.solutions: print(solution.score, solution.starts_at)

Source code in src/codex_services/booking/slot_master/scorer.py

class BookingScorer:
    """
    Evaluates engine solutions and returns an EngineResult with populated scores.

    Usage:
        scorer = BookingScorer(
            weights=ScoringWeights(preferred_resource_bonus=15.0),
            preferred_resource_ids=["3", "7"],
        )
        ranked = scorer.score(result)

        # Best solution by score (not just the earliest):
        print(ranked.best.score)
        print(ranked.best.to_display())

        # All solutions sorted by score (highest -> first):
        for solution in ranked.solutions:
            print(solution.score, solution.starts_at)
    """

    def __init__(
        self,
        weights: ScoringWeights | None = None,
        preferred_resource_ids: list[str] | None = None,
    ) -> None:
        """
        Args:
            weights: Criteria weights. None = ScoringWeights() with defaults.
            preferred_resource_ids: List of IDs for preferred resources.
                                    Must be strings (str(resource.pk)).
                                    On match -> preferred_resource_bonus applied.
        """
        self.weights = weights or ScoringWeights()
        self.preferred_ids: set[str] = set(preferred_resource_ids or [])

    def score(self, result: EngineResult) -> EngineResult:
        """
        Populates a score for each solution and returns a re-sorted EngineResult.

        Sorting: descending by score (best = first = result.best).
        Solutions with the same score are sorted by starts_at (earlier = better).

        Args:
            result: EngineResult from ChainFinder.find().

        Returns:
            New EngineResult (using frozen=True -> model_copy) with populated scores
            and solutions sorted by score DESC.
            If result is empty, it is returned unchanged.
        """
        if not result.solutions:
            return result

        scored = [self._score_solution(s) for s in result.solutions]
        # Sorting: score DESC, then starts_at ASC (earlier is better on equal score)
        scored.sort(key=lambda s: (-s.score, s.starts_at))

        return result.model_copy(update={"solutions": scored})

    def _score_solution(self, solution: BookingChainSolution) -> BookingChainSolution:
        """Calculates score for a single solution."""
        score = 0.0
        w = self.weights

        # --- Bonus for preferred resource ---
        if self.preferred_ids:
            for item in solution.items:
                if item.resource_id in self.preferred_ids:
                    score += w.preferred_resource_bonus

        # --- Bonus for one resource carrying out multiple services ---
        if w.same_resource_bonus > 0 and len(solution.items) > 1:
            resource_counts: dict[str, int] = {}
            for item in solution.items:
                resource_counts[item.resource_id] = resource_counts.get(item.resource_id, 0) + 1
            # Bonus for each pair of services with the same resource
            for count in resource_counts.values():
                if count > 1:
                    score += w.same_resource_bonus * (count - 1)

        # --- Bonus for chain compactness (minimal idle time) ---
        if w.min_idle_bonus_per_hour > 0 and len(solution.items) > 1:
            # Idle time = span - total service duration
            total_service_minutes = sum(item.duration_minutes for item in solution.items)
            idle_minutes = solution.span_minutes - total_service_minutes
            idle_hours = idle_minutes / 60.0
            # Less idle = higher bonus (max for 0 idle)
            max_idle_hours = solution.span_minutes / 60.0
            compactness = 1.0 - (idle_hours / max_idle_hours) if max_idle_hours > 0 else 1.0
            score += w.min_idle_bonus_per_hour * compactness * max_idle_hours

        # --- Penalty for late start (encourages early slots) ---
        if w.early_slot_penalty_per_hour > 0:
            # Calculate hours from the start of the day (00:00) to starts_at
            starts_hour = solution.starts_at.hour + solution.starts_at.minute / 60.0
            score -= w.early_slot_penalty_per_hour * starts_hour

        return solution.model_copy(update={"score": round(score, 4)})

Functions

__init__(weights=None, preferred_resource_ids=None)

Parameters:

Name	Type	Description	Default
weights	ScoringWeights | None	Criteria weights. None = ScoringWeights() with defaults.	None
preferred_resource_ids	list[str] | None	List of IDs for preferred resources. Must be strings (str(resource.pk)). On match -> preferred_resource_bonus applied.	None

Source code in src/codex_services/booking/slot_master/scorer.py

def __init__(
    self,
    weights: ScoringWeights | None = None,
    preferred_resource_ids: list[str] | None = None,
) -> None:
    """
    Args:
        weights: Criteria weights. None = ScoringWeights() with defaults.
        preferred_resource_ids: List of IDs for preferred resources.
                                Must be strings (str(resource.pk)).
                                On match -> preferred_resource_bonus applied.
    """
    self.weights = weights or ScoringWeights()
    self.preferred_ids: set[str] = set(preferred_resource_ids or [])

score(result)

Populates a score for each solution and returns a re-sorted EngineResult.

Sorting: descending by score (best = first = result.best). Solutions with the same score are sorted by starts_at (earlier = better).

Parameters:

Name	Type	Description	Default
result	EngineResult	EngineResult from ChainFinder.find().	required

Returns:

Type	Description
EngineResult	New EngineResult (using frozen=True -> model_copy) with populated scores
EngineResult	and solutions sorted by score DESC.
EngineResult	If result is empty, it is returned unchanged.

Source code in src/codex_services/booking/slot_master/scorer.py

def score(self, result: EngineResult) -> EngineResult:
    """
    Populates a score for each solution and returns a re-sorted EngineResult.

    Sorting: descending by score (best = first = result.best).
    Solutions with the same score are sorted by starts_at (earlier = better).

    Args:
        result: EngineResult from ChainFinder.find().

    Returns:
        New EngineResult (using frozen=True -> model_copy) with populated scores
        and solutions sorted by score DESC.
        If result is empty, it is returned unchanged.
    """
    if not result.solutions:
        return result

    scored = [self._score_solution(s) for s in result.solutions]
    # Sorting: score DESC, then starts_at ASC (earlier is better on equal score)
    scored.sort(key=lambda s: (-s.score, s.starts_at))

    return result.model_copy(update={"solutions": scored})

ScoringWeights dataclass

Weights for solution evaluation criteria. Configurable per project.

All weights are additive: total score = sum of applicable bonuses. Higher score = more preferred solution.

Fields

preferred_resource_bonus (float): Bonus for each service with a preferred resource. Passed via BookingScorer(preferred_resource_ids=[...]). Example: client always visits Anya -> preferred_resource_ids=["3"].

same_resource_bonus (float): Bonus if one resource performs multiple services. Reduces the number of times the client must switch resources.

min_idle_bonus_per_hour (float): Bonus for each hour of minimized idle time between services. Encourages compact chains.

early_slot_penalty_per_hour (float): Penalty for each hour from the start of the workday to the first service. Encourages earlier slots (less index = better). Negative value is not needed here — subtracted automatically.

Example (encourage early times AND preferred resource): ScoringWeights( preferred_resource_bonus=20.0, early_slot_penalty_per_hour=1.0, )

Source code in src/codex_services/booking/slot_master/scorer.py

@dataclass
class ScoringWeights:
    """
    Weights for solution evaluation criteria. Configurable per project.

    All weights are additive: total score = sum of applicable bonuses.
    Higher score = more preferred solution.

    Fields:
        preferred_resource_bonus (float):
            Bonus for each service with a preferred resource.
            Passed via BookingScorer(preferred_resource_ids=[...]).
            Example: client always visits Anya -> preferred_resource_ids=["3"].

        same_resource_bonus (float):
            Bonus if one resource performs multiple services.
            Reduces the number of times the client must switch resources.

        min_idle_bonus_per_hour (float):
            Bonus for each hour of minimized idle time between services.
            Encourages compact chains.

        early_slot_penalty_per_hour (float):
            Penalty for each hour from the start of the workday to the first service.
            Encourages earlier slots (less index = better).
            Negative value is not needed here — subtracted automatically.

    Example (encourage early times AND preferred resource):
        ScoringWeights(
            preferred_resource_bonus=20.0,
            early_slot_penalty_per_hour=1.0,
        )
    """

    preferred_resource_bonus: float = 10.0
    same_resource_bonus: float = 5.0
    min_idle_bonus_per_hour: float = 2.0
    early_slot_penalty_per_hour: float = 0.0

Functions

find_nearest_slots(request_data, get_availability_fn, *, search_from=None, search_days=60, step_minutes=30)

Search for the nearest available booking slots across multiple days.

Iterates days starting from search_from until a day with solutions is found.

Parameters:

Name	Type	Description	Default
request_data	dict[str, Any]	Dict matching :class:~codex_services.booking.slot_master.BookingEngineRequest.	required
get_availability_fn	Callable[[date], list[dict[str, Any]]]	Callable (date) -> list[dict]. Called for each checked day. Each dict in the returned list must match :class:~codex_services.booking.slot_master.MasterAvailability.	required
search_from	date | None	Start date (inclusive). Defaults to :func:~datetime.date.today.	None
search_days	int	Max number of days to search. Defaults to 60.	60
step_minutes	int	Slot grid step in minutes. Defaults to 30.	30

Returns:

Name	Type	Description
Serialised	dict[str, Any] | None	class:~codex_services.booking.slot_master.EngineResult dict for the first
	dict[str, Any] | None	matching day, or None if no slots were found within search_days.

Example::

def get_availability(d):
    # return your per-day availability data as list of dicts
    return [...]

result = find_nearest_slots(
    request_data={...},
    get_availability_fn=get_availability,
    search_from=date.today(),
)
if result:
    print(result["solutions"][0]["items"][0]["start_time"])

Source code in src/codex_services/booking/slot_master/api.py

def find_nearest_slots(
    request_data: dict[str, Any],
    get_availability_fn: Callable[[date], list[dict[str, Any]]],
    *,
    search_from: date | None = None,
    search_days: int = 60,
    step_minutes: int = 30,
) -> dict[str, Any] | None:
    """Search for the nearest available booking slots across multiple days.

    Iterates days starting from *search_from* until a day with solutions is found.

    Args:
        request_data: Dict matching :class:`~codex_services.booking.slot_master.BookingEngineRequest`.
        get_availability_fn: Callable ``(date) -> list[dict]``. Called for each checked
            day. Each dict in the returned list must match
            :class:`~codex_services.booking.slot_master.MasterAvailability`.
        search_from: Start date (inclusive). Defaults to :func:`~datetime.date.today`.
        search_days: Max number of days to search. Defaults to 60.
        step_minutes: Slot grid step in minutes. Defaults to 30.

    Returns:
        Serialised :class:`~codex_services.booking.slot_master.EngineResult` dict for the first
        matching day, or ``None`` if no slots were found within *search_days*.

    Example::

        def get_availability(d):
            # return your per-day availability data as list of dicts
            return [...]

        result = find_nearest_slots(
            request_data={...},
            get_availability_fn=get_availability,
            search_from=date.today(),
        )
        if result:
            print(result["solutions"][0]["items"][0]["start_time"])
    """
    search_from = search_from or date.today()
    request = BookingEngineRequest.model_validate(request_data)

    def _get_avail(d: date) -> dict[str, MasterAvailability]:
        return _parse_availability(get_availability_fn(d))

    result = ChainFinder(step_minutes).find_nearest(
        request, _get_avail, search_from=search_from, search_days=search_days
    )

    return result.model_dump(mode="json") if result.has_solutions else None

find_slots(request_data, resources_availability, *, step_minutes=30, max_solutions=50, scoring_weights=None, preferred_resource_ids=None)

Find available booking slots for one day.

Validates input via Pydantic, runs :class:ChainFinder, optionally scores results.

Parameters:

Name	Type	Description	Default
request_data	dict[str, Any]	Dict matching :class:~codex_services.booking.slot_master.BookingEngineRequest. Required: service_requests, booking_date.	required
resources_availability	list[dict[str, Any]]	List of dicts matching :class:~codex_services.booking.slot_master.MasterAvailability. Each item must contain resource_id and free_windows.	required
step_minutes	int	Slot grid step in minutes. Defaults to 30.	30
max_solutions	int	Max solutions returned by the engine. Defaults to 50.	50
scoring_weights	dict[str, Any] | None	Optional dict with :class:~codex_services.booking.slot_master.scorer.ScoringWeights fields to rank solutions. Supported keys: preferred_resource_bonus, same_resource_bonus, min_idle_bonus_per_hour, early_slot_penalty_per_hour.	None
preferred_resource_ids	list[str] | None	Resource IDs to boost in scoring. Only effective when scoring_weights is provided.	None

Returns:

Name	Type	Description
Serialised	dict[str, Any]	class:~codex_services.booking.slot_master.EngineResult dict.
	dict[str, Any]	Key "solutions" is a list of booking chain dicts.
	dict[str, Any]	Returns an empty list under "solutions" if no slots are available.

Raises:

Type	Description
ValidationError	If request_data or resources_availability do not match the expected schema.
NotImplementedError	If the request uses an unsupported mode (e.g. MULTI_DAY or group bookings).

Example::

result = find_slots(
    request_data={
        "service_requests": [
            {"service_id": "s1", "duration_minutes": 60,
             "possible_resource_ids": ["m1"]},
        ],
        "booking_date": "2024-05-15",
    },
    resources_availability=[
        {"resource_id": "m1",
         "free_windows": [["2024-05-15T09:00:00", "2024-05-15T18:00:00"]]},
    ],
)
for chain in result["solutions"]:
    print(chain["items"][0]["start_time"])

Source code in src/codex_services/booking/slot_master/api.py

def find_slots(
    request_data: dict[str, Any],
    resources_availability: list[dict[str, Any]],
    *,
    step_minutes: int = 30,
    max_solutions: int = 50,
    scoring_weights: dict[str, Any] | None = None,
    preferred_resource_ids: list[str] | None = None,
) -> dict[str, Any]:
    """Find available booking slots for one day.

    Validates input via Pydantic, runs :class:`ChainFinder`, optionally scores results.

    Args:
        request_data: Dict matching :class:`~codex_services.booking.slot_master.BookingEngineRequest`.
            Required: ``service_requests``, ``booking_date``.
        resources_availability: List of dicts matching
            :class:`~codex_services.booking.slot_master.MasterAvailability`.
            Each item must contain ``resource_id`` and ``free_windows``.
        step_minutes: Slot grid step in minutes. Defaults to 30.
        max_solutions: Max solutions returned by the engine. Defaults to 50.
        scoring_weights: Optional dict with :class:`~codex_services.booking.slot_master.scorer.ScoringWeights`
            fields to rank solutions. Supported keys: ``preferred_resource_bonus``,
            ``same_resource_bonus``, ``min_idle_bonus_per_hour``,
            ``early_slot_penalty_per_hour``.
        preferred_resource_ids: Resource IDs to boost in scoring.
            Only effective when ``scoring_weights`` is provided.

    Returns:
        Serialised :class:`~codex_services.booking.slot_master.EngineResult` dict.
        Key ``"solutions"`` is a list of booking chain dicts.
        Returns an empty list under ``"solutions"`` if no slots are available.

    Raises:
        pydantic.ValidationError: If ``request_data`` or ``resources_availability``
            do not match the expected schema.
        NotImplementedError: If the request uses an unsupported mode
            (e.g. MULTI_DAY or group bookings).

    Example::

        result = find_slots(
            request_data={
                "service_requests": [
                    {"service_id": "s1", "duration_minutes": 60,
                     "possible_resource_ids": ["m1"]},
                ],
                "booking_date": "2024-05-15",
            },
            resources_availability=[
                {"resource_id": "m1",
                 "free_windows": [["2024-05-15T09:00:00", "2024-05-15T18:00:00"]]},
            ],
        )
        for chain in result["solutions"]:
            print(chain["items"][0]["start_time"])
    """
    request = BookingEngineRequest.model_validate(request_data)
    availability = _parse_availability(resources_availability)

    result: EngineResult = ChainFinder(step_minutes).find(request, availability, max_solutions=max_solutions)

    if scoring_weights is not None:
        weights = ScoringWeights(**scoring_weights)
        scorer = BookingScorer(weights=weights, preferred_resource_ids=preferred_resource_ids)
        result = scorer.score(result)

    return result.model_dump(mode="json")