Skip to content

codex_services.booking

booking

codex_services.booking: Unified booking engine library.

Provides different booking strategies (slot_master, room, etc.) with shared utilities and base DTOs.

Classes

SlotCalculator

Slot generator inside a free time window (sliding window).

Does not depend on ORM. Works with datetime objects.

Examples:

calc = SlotCalculator(step_minutes=30)

# Get all possible slots in a window from 9:00 to 12:00 for a 60 min service:
slots = calc.find_slots_in_window(
    window_start=datetime(2024, 5, 10, 9, 0),
    window_end=datetime(2024, 5, 10, 12, 0),
    duration_minutes=60,
)
# -> [datetime(2024,5,10,9,0), datetime(2024,5,10,9,30), ...]

# Get free windows from a working day:
windows = calc.merge_free_windows(
    work_start=datetime(2024,5,10,9,0),
    work_end=datetime(2024,5,10,18,0),
    busy_intervals=[(datetime(2024,5,10,10,0), datetime(2024,5,10,11,0))],
)
# -> [(9:00, 10:00), (11:10, 18:00)]
Source code in src/codex_services/booking/_shared/calculator.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
class SlotCalculator:
    """
    Slot generator inside a free time window (sliding window).

    Does not depend on ORM. Works with datetime objects.

    Examples:
        ```python
        calc = SlotCalculator(step_minutes=30)

        # Get all possible slots in a window from 9:00 to 12:00 for a 60 min service:
        slots = calc.find_slots_in_window(
            window_start=datetime(2024, 5, 10, 9, 0),
            window_end=datetime(2024, 5, 10, 12, 0),
            duration_minutes=60,
        )
        # -> [datetime(2024,5,10,9,0), datetime(2024,5,10,9,30), ...]
        ```

        ```python
        # Get free windows from a working day:
        windows = calc.merge_free_windows(
            work_start=datetime(2024,5,10,9,0),
            work_end=datetime(2024,5,10,18,0),
            busy_intervals=[(datetime(2024,5,10,10,0), datetime(2024,5,10,11,0))],
        )
        # -> [(9:00, 10:00), (11:10, 18:00)]
        ```
    """

    def __init__(self, step_minutes: int = 30) -> None:
        """
        Args:
            step_minutes: Grid slot step in minutes. Defaults to 30.
                          Determines the interval at which slots are offered.
        """
        if step_minutes <= 0:
            raise ValueError(f"step_minutes должен быть > 0, получен: {step_minutes}")
        self.step_minutes = step_minutes
        self._step_delta = timedelta(minutes=step_minutes)

    def find_slots_in_window(
        self,
        window_start: datetime,
        window_end: datetime,
        duration_minutes: int,
        min_start: datetime | None = None,
        grid_anchor: datetime | None = None,
    ) -> list[datetime]:
        """
        Returns a list of possible start times within the window.

        Algorithm: sliding window with a step of self.step_minutes.
        Each candidate is checked: does the service [slot, slot+duration]
        fit within the end of the window.

        Args:
            window_start: Start of the free time window.
            window_end: End of the free time window.
            duration_minutes: Duration of the service in minutes.
            min_start: Minimum acceptable start time (e.g., "no earlier than
                       15 minutes from the current time"). None = no limit.
            grid_anchor: Anchor for grid alignment (e.g., shifts start).
                         If provided, slots will be aligned to (grid_anchor + N * step).

        Returns:
            List of datetime — possible service start moments.
            Empty list if the service does not fit in the window.

        Example:
            # Window 9:00-11:00, service 60 min, step 30 min:
            # -> [9:00, 9:30, 10:00]  (10:30 + 60 min = 11:30, does not fit)
        """
        duration_delta = timedelta(minutes=duration_minutes)

        # Быстрая проверка: окно в принципе достаточно для услуги
        if window_end - window_start < duration_delta:
            return []

        slots: list[datetime] = []
        current = window_start

        # Если есть min_start — двигаем указатель к ближайшему шагу сетки
        if min_start and current < min_start:
            current = min_start

        # Если задан якорь — выравниваем текущую позицию по сетке относительно якоря
        # Если якоря нет — по умолчанию выравниваем по началу окна (обратная совместимость)
        anchor = grid_anchor or window_start
        current = self._align_to_grid(current, anchor)

        while current + duration_delta <= window_end:
            slots.append(current)
            current += self._step_delta

        return slots

    def merge_free_windows(
        self,
        work_start: datetime,
        work_end: datetime,
        busy_intervals: list[tuple[datetime, datetime]],
        break_interval: tuple[datetime, datetime] | None = None,
        buffer_minutes: int = 0,
        min_duration_minutes: int = 0,
    ) -> list[tuple[datetime, datetime]]:
        """
        Calculates a list of free windows from a working day.

        Subtracts from the interval [work_start, work_end]:
        - busy_intervals (booked appointments)
        - break_interval (resource's break)
        - buffer_minutes (buffer after each booked interval)

        Args:
            work_start: Start of the resource's working day.
            work_end: End of the resource's working day.
            busy_intervals: List of busy segments [(start, end), ...].
            break_interval: Resource's break (lunch), tuple (start, end) or None.
            buffer_minutes: Buffer in minutes added after each booked appointment.
            min_duration_minutes: Minimum window length. Windows shorter than this value
                                  are discarded as "junk".

        Returns:
            List of free windows [(start, end), ...] sorted by time.

        Example:
            ```python
            # Work day 9:00-18:00, booked 10:00-11:00, lunch 13:00-14:00
            windows = calc.merge_free_windows(
                work_start=dt(9,0), work_end=dt(18,0),
                busy_intervals=[(dt(10,0), dt(11,0))],
                break_interval=(dt(13,0), dt(14,0)),
                buffer_minutes=10,
            )
            # -> [(9:00, 10:00), (11:10, 13:00), (14:00, 18:00)]
            ```
        """
        buffer_delta = timedelta(minutes=buffer_minutes)

        # Collect all "busy" intervals: appointments + break
        blocked: list[tuple[datetime, datetime]] = []
        for b_start, b_end in busy_intervals:
            # Apply buffer after the booked appointment
            blocked.append((b_start, b_end + buffer_delta))

        if break_interval:
            blocked.append(break_interval)

        # Sort and merge overlapping intervals
        blocked = self._merge_intervals(blocked)

        # Calculate free windows
        free_windows: list[tuple[datetime, datetime]] = []
        current_ptr = work_start

        for b_start, b_end in blocked:
            # Cut off segments outside the working day
            b_start = max(b_start, work_start)
            b_end = min(b_end, work_end)

            if b_start > current_ptr:
                # Junk window filter
                if min_duration_minutes > 0:
                    duration = (b_start - current_ptr).total_seconds() / 60
                    if duration < min_duration_minutes:
                        current_ptr = max(current_ptr, b_end)
                        continue

                free_windows.append((current_ptr, b_start))

            current_ptr = max(current_ptr, b_end)

        # Remaining time after the last busy block
        if current_ptr < work_end:
            if min_duration_minutes > 0:
                duration = (work_end - current_ptr).total_seconds() / 60
                if duration >= min_duration_minutes:
                    free_windows.append((current_ptr, work_end))
            else:
                free_windows.append((current_ptr, work_end))

        return free_windows

    def find_gaps(
        self,
        free_windows: list[tuple[datetime, datetime]],
        min_gap_minutes: int,
    ) -> list[tuple[datetime, datetime, int]]:
        """
        Finds all free windows of minimum length in the resource's schedule.

        Used for:
            - Load analysis: how much free time the resource has
            - Notifications: resource is free for N minutes -> offer to user

        Args:
            free_windows: Free windows from ResourceAvailability.free_windows
            min_gap_minutes: Minimum window length in minutes.

        Returns:
            List (window_start, window_end, duration_minutes) — windows
            longer than min_gap_minutes, sorted by start time.

        Examples:
            ```python
            # Search for windows >= 60 minutes:
            gaps = calc.find_gaps(free_windows, min_gap_minutes=60)
            # -> [(9:00, 10:00, 60), (11:30, 14:00, 150), (16:00, 18:00, 120)]
            ```
        """
        result: list[tuple[datetime, datetime, int]] = []

        for w_start, w_end in free_windows:
            duration = int((w_end - w_start).total_seconds() / 60)
            if duration >= min_gap_minutes:
                result.append((w_start, w_end, duration))

        result.sort(key=lambda x: x[0])
        return result

    def split_window_by_service(
        self,
        window_start: datetime,
        window_end: datetime,
        service_start: datetime,
        service_end: datetime,
    ) -> list[tuple[datetime, datetime]]:
        """
        Splits a free window into parts around a booked service segment.

        Used for dynamic calculation: "if we put a service here —
        what windows will remain free for the next ones?"

        Args:
            window_start: Start of the free window.
            window_end: End of the free window.
            service_start: Start of the booked segment (must be inside the window).
            service_end: End of the booked segment (must be inside the window).

        Returns:
            List of remaining free windows (0, 1, or 2 elements).

        Example:
            # Window 9:00-18:00, service 11:00-12:00:
            split_window_by_service(9:00, 18:00, 11:00, 12:00)
            # -> [(9:00, 11:00), (12:00, 18:00)]

            # Service at the start of the window:
            split_window_by_service(9:00, 18:00, 9:00, 11:00)
            # -> [(11:00, 18:00)]
        """
        remaining: list[tuple[datetime, datetime]] = []

        # Part before the service
        if service_start > window_start:
            remaining.append((window_start, service_start))

        # Part after the service
        if service_end < window_end:
            remaining.append((service_end, window_end))

        return remaining

    # ---------------------------------------------------------------------------
    # Internal helper methods
    # ---------------------------------------------------------------------------

    def _merge_intervals(self, intervals: list[tuple[datetime, datetime]]) -> list[tuple[datetime, datetime]]:
        """
        Merges overlapping or adjacent intervals.
        Returns a sorted list of non-overlapping intervals.

        Internal method. Used in merge_free_windows.
        """
        if not intervals:
            return []

        sorted_intervals = sorted(intervals, key=lambda x: x[0])
        merged: list[tuple[datetime, datetime]] = [sorted_intervals[0]]

        for start, end in sorted_intervals[1:]:
            last_start, last_end = merged[-1]
            if start <= last_end:
                # Overlap — expand current block
                merged[-1] = (last_start, max(last_end, end))
            else:
                merged.append((start, end))

        return merged

    def _align_to_grid(self, target: datetime, grid_origin: datetime) -> datetime:
        """
        Aligns target to the nearest grid step relative to grid_origin.
        Returns the first grid moment >= target.

        Internal method. Used to align min_start to the grid.

        Example (step 30 min, origin 9:00, target 9:17):
            -> 9:30  (nearest step >= 9:17)
        """
        delta_seconds = (target - grid_origin).total_seconds()
        step_seconds = self._step_delta.total_seconds()

        if delta_seconds <= 0:
            return grid_origin

        # Number of full steps
        full_steps = int(delta_seconds / step_seconds)
        aligned = grid_origin + timedelta(seconds=full_steps * step_seconds)

        if aligned < target:
            aligned += self._step_delta

        return aligned
Functions
__init__(step_minutes=30)

Parameters:

Name Type Description Default
step_minutes int

Grid slot step in minutes. Defaults to 30. Determines the interval at which slots are offered.

 30
Source code in src/codex_services/booking/_shared/calculator.py 
46
47
48
49
50
51
52
53
54
55
def __init__(self, step_minutes: int = 30) -> None:
    """
    Args:
        step_minutes: Grid slot step in minutes. Defaults to 30.
                      Determines the interval at which slots are offered.
    """
    if step_minutes <= 0:
        raise ValueError(f"step_minutes должен быть > 0, получен: {step_minutes}")
    self.step_minutes = step_minutes
    self._step_delta = timedelta(minutes=step_minutes)
find_slots_in_window(window_start, window_end, duration_minutes, min_start=None, grid_anchor=None)

Returns a list of possible start times within the window.

Algorithm: sliding window with a step of self.step_minutes. Each candidate is checked: does the service [slot, slot+duration] fit within the end of the window.

Parameters:

Name Type Description Default
window_start datetime

Start of the free time window.

 required
window_end datetime

End of the free time window.

 required
duration_minutes int

Duration of the service in minutes.

 required
min_start datetime | None

Minimum acceptable start time (e.g., "no earlier than 15 minutes from the current time"). None = no limit.

 None
grid_anchor datetime | None

Anchor for grid alignment (e.g., shifts start). If provided, slots will be aligned to (grid_anchor + N * step).

 None

Returns:

Type Description
list[datetime]

List of datetime — possible service start moments.
list[datetime]

Empty list if the service does not fit in the window.
Example
Window 9:00-11:00, service 60 min, step 30 min:
-> [9:00, 9:30, 10:00] (10:30 + 60 min = 11:30, does not fit)
Source code in src/codex_services/booking/_shared/calculator.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def find_slots_in_window(
    self,
    window_start: datetime,
    window_end: datetime,
    duration_minutes: int,
    min_start: datetime | None = None,
    grid_anchor: datetime | None = None,
) -> list[datetime]:
    """
    Returns a list of possible start times within the window.

    Algorithm: sliding window with a step of self.step_minutes.
    Each candidate is checked: does the service [slot, slot+duration]
    fit within the end of the window.

    Args:
        window_start: Start of the free time window.
        window_end: End of the free time window.
        duration_minutes: Duration of the service in minutes.
        min_start: Minimum acceptable start time (e.g., "no earlier than
                   15 minutes from the current time"). None = no limit.
        grid_anchor: Anchor for grid alignment (e.g., shifts start).
                     If provided, slots will be aligned to (grid_anchor + N * step).

    Returns:
        List of datetime — possible service start moments.
        Empty list if the service does not fit in the window.

    Example:
        # Window 9:00-11:00, service 60 min, step 30 min:
        # -> [9:00, 9:30, 10:00]  (10:30 + 60 min = 11:30, does not fit)
    """
    duration_delta = timedelta(minutes=duration_minutes)

    # Быстрая проверка: окно в принципе достаточно для услуги
    if window_end - window_start < duration_delta:
        return []

    slots: list[datetime] = []
    current = window_start

    # Если есть min_start — двигаем указатель к ближайшему шагу сетки
    if min_start and current < min_start:
        current = min_start

    # Если задан якорь — выравниваем текущую позицию по сетке относительно якоря
    # Если якоря нет — по умолчанию выравниваем по началу окна (обратная совместимость)
    anchor = grid_anchor or window_start
    current = self._align_to_grid(current, anchor)

    while current + duration_delta <= window_end:
        slots.append(current)
        current += self._step_delta

    return slots
merge_free_windows(work_start, work_end, busy_intervals, break_interval=None, buffer_minutes=0, min_duration_minutes=0)

Calculates a list of free windows from a working day.

Subtracts from the interval [work_start, work_end]: - busy_intervals (booked appointments) - break_interval (resource's break) - buffer_minutes (buffer after each booked interval)

Parameters:

Name Type Description Default
work_start datetime

Start of the resource's working day.

 required
work_end datetime

End of the resource's working day.

 required
busy_intervals list[tuple[datetime, datetime]]

List of busy segments [(start, end), ...].

 required
break_interval tuple[datetime, datetime] | None

Resource's break (lunch), tuple (start, end) or None.

 None
buffer_minutes int

Buffer in minutes added after each booked appointment.

 0
min_duration_minutes int

Minimum window length. Windows shorter than this value are discarded as "junk".

 0

Returns:

Type Description
list[tuple[datetime, datetime]]

List of free windows [(start, end), ...] sorted by time.
Example 
# Work day 9:00-18:00, booked 10:00-11:00, lunch 13:00-14:00
windows = calc.merge_free_windows(
    work_start=dt(9,0), work_end=dt(18,0),
    busy_intervals=[(dt(10,0), dt(11,0))],
    break_interval=(dt(13,0), dt(14,0)),
    buffer_minutes=10,
)
# -> [(9:00, 10:00), (11:10, 13:00), (14:00, 18:00)]
Source code in src/codex_services/booking/_shared/calculator.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def merge_free_windows(
    self,
    work_start: datetime,
    work_end: datetime,
    busy_intervals: list[tuple[datetime, datetime]],
    break_interval: tuple[datetime, datetime] | None = None,
    buffer_minutes: int = 0,
    min_duration_minutes: int = 0,
) -> list[tuple[datetime, datetime]]:
    """
    Calculates a list of free windows from a working day.

    Subtracts from the interval [work_start, work_end]:
    - busy_intervals (booked appointments)
    - break_interval (resource's break)
    - buffer_minutes (buffer after each booked interval)

    Args:
        work_start: Start of the resource's working day.
        work_end: End of the resource's working day.
        busy_intervals: List of busy segments [(start, end), ...].
        break_interval: Resource's break (lunch), tuple (start, end) or None.
        buffer_minutes: Buffer in minutes added after each booked appointment.
        min_duration_minutes: Minimum window length. Windows shorter than this value
                              are discarded as "junk".

    Returns:
        List of free windows [(start, end), ...] sorted by time.

    Example:
        ```python
        # Work day 9:00-18:00, booked 10:00-11:00, lunch 13:00-14:00
        windows = calc.merge_free_windows(
            work_start=dt(9,0), work_end=dt(18,0),
            busy_intervals=[(dt(10,0), dt(11,0))],
            break_interval=(dt(13,0), dt(14,0)),
            buffer_minutes=10,
        )
        # -> [(9:00, 10:00), (11:10, 13:00), (14:00, 18:00)]
        ```
    """
    buffer_delta = timedelta(minutes=buffer_minutes)

    # Collect all "busy" intervals: appointments + break
    blocked: list[tuple[datetime, datetime]] = []
    for b_start, b_end in busy_intervals:
        # Apply buffer after the booked appointment
        blocked.append((b_start, b_end + buffer_delta))

    if break_interval:
        blocked.append(break_interval)

    # Sort and merge overlapping intervals
    blocked = self._merge_intervals(blocked)

    # Calculate free windows
    free_windows: list[tuple[datetime, datetime]] = []
    current_ptr = work_start

    for b_start, b_end in blocked:
        # Cut off segments outside the working day
        b_start = max(b_start, work_start)
        b_end = min(b_end, work_end)

        if b_start > current_ptr:
            # Junk window filter
            if min_duration_minutes > 0:
                duration = (b_start - current_ptr).total_seconds() / 60
                if duration < min_duration_minutes:
                    current_ptr = max(current_ptr, b_end)
                    continue

            free_windows.append((current_ptr, b_start))

        current_ptr = max(current_ptr, b_end)

    # Remaining time after the last busy block
    if current_ptr < work_end:
        if min_duration_minutes > 0:
            duration = (work_end - current_ptr).total_seconds() / 60
            if duration >= min_duration_minutes:
                free_windows.append((current_ptr, work_end))
        else:
            free_windows.append((current_ptr, work_end))

    return free_windows
find_gaps(free_windows, min_gap_minutes)

Finds all free windows of minimum length in the resource's schedule.

Used for
  • Load analysis: how much free time the resource has
  • Notifications: resource is free for N minutes -> offer to user

Parameters:

Name Type Description Default
free_windows list[tuple[datetime, datetime]]

Free windows from ResourceAvailability.free_windows

 required
min_gap_minutes int

Minimum window length in minutes.

 required

Returns:

Type Description
list[tuple[datetime, datetime, int]]

List (window_start, window_end, duration_minutes) — windows
list[tuple[datetime, datetime, int]]

longer than min_gap_minutes, sorted by start time.

Examples:

# Search for windows >= 60 minutes:
gaps = calc.find_gaps(free_windows, min_gap_minutes=60)
# -> [(9:00, 10:00, 60), (11:30, 14:00, 150), (16:00, 18:00, 120)]
Source code in src/codex_services/booking/_shared/calculator.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def find_gaps(
    self,
    free_windows: list[tuple[datetime, datetime]],
    min_gap_minutes: int,
) -> list[tuple[datetime, datetime, int]]:
    """
    Finds all free windows of minimum length in the resource's schedule.

    Used for:
        - Load analysis: how much free time the resource has
        - Notifications: resource is free for N minutes -> offer to user

    Args:
        free_windows: Free windows from ResourceAvailability.free_windows
        min_gap_minutes: Minimum window length in minutes.

    Returns:
        List (window_start, window_end, duration_minutes) — windows
        longer than min_gap_minutes, sorted by start time.

    Examples:
        ```python
        # Search for windows >= 60 minutes:
        gaps = calc.find_gaps(free_windows, min_gap_minutes=60)
        # -> [(9:00, 10:00, 60), (11:30, 14:00, 150), (16:00, 18:00, 120)]
        ```
    """
    result: list[tuple[datetime, datetime, int]] = []

    for w_start, w_end in free_windows:
        duration = int((w_end - w_start).total_seconds() / 60)
        if duration >= min_gap_minutes:
            result.append((w_start, w_end, duration))

    result.sort(key=lambda x: x[0])
    return result
split_window_by_service(window_start, window_end, service_start, service_end)

Splits a free window into parts around a booked service segment.

Used for dynamic calculation: "if we put a service here — what windows will remain free for the next ones?"

Parameters:

Name Type Description Default
window_start datetime

Start of the free window.

 required
window_end datetime

End of the free window.

 required
service_start datetime

Start of the booked segment (must be inside the window).

 required
service_end datetime

End of the booked segment (must be inside the window).

 required

Returns:

Type Description
list[tuple[datetime, datetime]]

List of remaining free windows (0, 1, or 2 elements).
Example
Window 9:00-18:00, service 11:00-12:00:

split_window_by_service(9:00, 18:00, 11:00, 12:00)

-> [(9:00, 11:00), (12:00, 18:00)]
Service at the start of the window:

split_window_by_service(9:00, 18:00, 9:00, 11:00)

-> [(11:00, 18:00)]
Source code in src/codex_services/booking/_shared/calculator.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def split_window_by_service(
    self,
    window_start: datetime,
    window_end: datetime,
    service_start: datetime,
    service_end: datetime,
) -> list[tuple[datetime, datetime]]:
    """
    Splits a free window into parts around a booked service segment.

    Used for dynamic calculation: "if we put a service here —
    what windows will remain free for the next ones?"

    Args:
        window_start: Start of the free window.
        window_end: End of the free window.
        service_start: Start of the booked segment (must be inside the window).
        service_end: End of the booked segment (must be inside the window).

    Returns:
        List of remaining free windows (0, 1, or 2 elements).

    Example:
        # Window 9:00-18:00, service 11:00-12:00:
        split_window_by_service(9:00, 18:00, 11:00, 12:00)
        # -> [(9:00, 11:00), (12:00, 18:00)]

        # Service at the start of the window:
        split_window_by_service(9:00, 18:00, 9:00, 11:00)
        # -> [(11:00, 18:00)]
    """
    remaining: list[tuple[datetime, datetime]] = []

    # Part before the service
    if service_start > window_start:
        remaining.append((window_start, service_start))

    # Part after the service
    if service_end < window_end:
        remaining.append((service_end, window_end))

    return remaining

BookingRequest

Bases: BaseDTO

Base booking request.

Fields

booking_date (date): Target date for the booking.

Source code in src/codex_services/booking/_shared/dto.py 
41
42
43
44
45
46
47
48
49
class BookingRequest(BaseDTO):
    """
    Base booking request.

    Fields:
        booking_date (date): Target date for the booking.
    """

    booking_date: date

BookingResult

Bases: BaseDTO

Base result — list of solutions.

Concrete booking types override the solutions field with a specific type.

Source code in src/codex_services/booking/_shared/dto.py 
72
73
74
75
76
77
78
79
80
81
82
class BookingResult(BaseDTO):
    """
    Base result — list of solutions.

    Concrete booking types override the `solutions` field with a specific type.
    """

    @property
    def has_solutions(self) -> bool:
        """Return True if at least one solution was found."""
        raise NotImplementedError("Subclasses must implement has_solutions")
Attributes
has_solutions property

Return True if at least one solution was found.

BookingSolution

Bases: BaseDTO

Base result — one booked slot.

Fields

resource_id (str): Assigned resource identifier. start_time (datetime): Scheduled start time. end_time (datetime): Scheduled end time.

Source code in src/codex_services/booking/_shared/dto.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class BookingSolution(BaseDTO):
    """
    Base result — one booked slot.

    Fields:
        resource_id (str): Assigned resource identifier.
        start_time (datetime): Scheduled start time.
        end_time (datetime): Scheduled end time.
    """

    resource_id: str
    start_time: datetime
    end_time: datetime

    @property
    def duration_minutes(self) -> int:
        """Calculate actual duration in minutes."""
        return int((self.end_time - self.start_time).total_seconds() / 60)
Attributes
duration_minutes property

Calculate actual duration in minutes.

ResourceAvailability

Bases: BaseDTO

Base availability of a resource (executor, room, equipment, etc.).

Fields

resource_id (str): Unique identifier of the resource. free_windows (list[tuple[datetime, datetime]]): Available time windows. Each tuple is (start, end) where start < end.

Source code in src/codex_services/booking/_shared/dto.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class ResourceAvailability(BaseDTO):
    """
    Base availability of a resource (executor, room, equipment, etc.).

    Fields:
        resource_id (str): Unique identifier of the resource.
        free_windows (list[tuple[datetime, datetime]]): Available time windows.
            Each tuple is (start, end) where start < end.
    """

    resource_id: str
    free_windows: list[tuple[datetime, datetime]] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_windows_order(self) -> "ResourceAvailability":
        """Validate that each free window is chronologically correct."""
        for start, end in self.free_windows:
            if start >= end:
                raise ValueError(f"Resource {self.resource_id}: start={start} >= end={end}")
        return self

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

Validate that each free window is chronologically correct.

Source code in src/codex_services/booking/_shared/dto.py 
29
30
31
32
33
34
35
@model_validator(mode="after")
def validate_windows_order(self) -> "ResourceAvailability":
    """Validate that each free window is chronologically correct."""
    for start, end in self.free_windows:
        if start >= end:
            raise ValueError(f"Resource {self.resource_id}: start={start} >= end={end}")
    return self

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 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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)

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 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
45
46
47
48
49
50
51
52
53
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 
55
56
57
58
59
60
61
62
63
64
65
66
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 
29
30
31
32
33
34
35
36
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 
32
33
34
35
36
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 
17
18
19
20
21
22
23
24
25
26
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 
20
21
22
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 
24
25
26
def get_break_interval(self, resource_id: str, target_date: date) -> tuple[datetime, datetime] | None:
    """Return (start, end) of break or None."""
    ...

BookingValidator

A set of correctness checks for booking data. Unaffected by the ORM — operates exclusively on DTOs.

Used by
  • ChainFinder: Ensuring found chains have no conflicts.
  • Adapter: Final verification before creating Appointment instances in the DB.
  • Tests: Isolated logic verification without Django.
Example

v = BookingValidator()

Check if a slot is free:

ok = v.is_slot_free( slot_start=datetime(2024,5,10,10,0), slot_end=datetime(2024,5,10,11,0), busy_intervals=[(datetime(2024,5,10,9,0), datetime(2024,5,10,9,30))], )

→ True (no overlap)
Check entire chain for conflicts:

ok = v.no_conflicts(solutions)

Source code in src/codex_services/booking/_shared/validators.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class BookingValidator:
    """
    A set of correctness checks for booking data.
    Unaffected by the ORM — operates exclusively on DTOs.

    Used by:
        - ChainFinder: Ensuring found chains have no conflicts.
        - Adapter: Final verification before creating Appointment instances in the DB.
        - Tests: Isolated logic verification without Django.

    Example:
        v = BookingValidator()

        # Check if a slot is free:
        ok = v.is_slot_free(
            slot_start=datetime(2024,5,10,10,0),
            slot_end=datetime(2024,5,10,11,0),
            busy_intervals=[(datetime(2024,5,10,9,0), datetime(2024,5,10,9,30))],
        )
        # → True (no overlap)

        # Check entire chain for conflicts:
        ok = v.no_conflicts(solutions)
    """

    def is_slot_free(
        self,
        slot_start: datetime,
        slot_end: datetime,
        busy_intervals: list[tuple[datetime, datetime]],
    ) -> bool:
        """
        Verifies that the slot [slot_start, slot_end) does not overlap
        with any of the busy intervals.

        Uses a "half-open" interval — [start, end). If slot_end == busy_start,
        it is NOT considered a conflict (adjacent slots are allowed).

        Args:
            slot_start: Start of the slot to check.
            slot_end: End of the slot to check.
            busy_intervals: List of busy intervals [(start, end), ...].

        Returns:
            True if the slot is free. False if there is an overlap.

        Example:
            # Busy 10:00-11:00. Requesting 10:30-11:30 → conflict:
            is_slot_free(10:30, 11:30, [(10:00, 11:00)]) → False

            # Requesting 11:00-12:00 → OK (adjacent slots):
            is_slot_free(11:00, 12:00, [(10:00, 11:00)]) → True
        """
        return all(not (slot_start < busy_end and slot_end > busy_start) for busy_start, busy_end in busy_intervals)

    def no_conflicts(
        self,
        solutions: list[BookingSolution],
    ) -> bool:
        """
        Verifies that there are no resource conflicts within a set of solutions.
        A resource cannot be occupied by two services simultaneously.

        Groups solutions by resource_id and checks each group for overlaps.
        Used by ChainFinder after assembling the chain for final verification.

        Args:
            solutions: List of BookingSolution objects (found slots).

        Returns:
            True if no conflicts exist. False if at least one resource is double-booked.

        Example:
            no_conflicts([
                SingleServiceSolution(resource_id="1", start=9:00, gap_end=10:10),
                SingleServiceSolution(resource_id="1", start=10:10, gap_end=11:10),
            ])
            # → True (slots are adjacent, no overlap)
        """
        by_resource: dict[str, list[BookingSolution]] = {}
        for sol in solutions:
            by_resource.setdefault(sol.resource_id, []).append(sol)

        for _resource_id, resource_solutions in by_resource.items():
            if len(resource_solutions) < 2:
                continue
            sorted_sols = sorted(resource_solutions, key=lambda s: s.start_time)
            for i in range(len(sorted_sols) - 1):
                current = sorted_sols[i]
                next_sol = sorted_sols[i + 1]
                # For solutions with gap_end_time, use it; otherwise use end_time
                block_end = getattr(current, "gap_end_time", current.end_time)
                if next_sol.start_time < block_end:
                    return False

        return True

    def solution_fits_in_windows(
        self,
        solution: BookingSolution,
        free_windows: list[tuple[datetime, datetime]],
    ) -> bool:
        """
        Verifies that a solution's slot fits entirely inside one of the
        resource's free windows.

        Args:
            solution: Found slot for a single service.
            free_windows: Resource's free windows (from ResourceAvailability).

        Returns:
            True if the slot fits perfectly inside one of the given free windows.
        """
        block_end = getattr(solution, "gap_end_time", solution.end_time)
        return any(solution.start_time >= w_start and block_end <= w_end for w_start, w_end in free_windows)
Functions
is_slot_free(slot_start, slot_end, busy_intervals)

Verifies that the slot [slot_start, slot_end) does not overlap with any of the busy intervals.

Uses a "half-open" interval — [start, end). If slot_end == busy_start, it is NOT considered a conflict (adjacent slots are allowed).

Parameters:

Name Type Description Default
slot_start datetime

Start of the slot to check.

 required
slot_end datetime

End of the slot to check.

 required
busy_intervals list[tuple[datetime, datetime]]

List of busy intervals [(start, end), ...].

 required

Returns:

Type Description
bool

True if the slot is free. False if there is an overlap.
Example
Busy 10:00-11:00. Requesting 10:30-11:30 → conflict:

is_slot_free(10:30, 11:30, [(10:00, 11:00)]) → False

Requesting 11:00-12:00 → OK (adjacent slots):

is_slot_free(11:00, 12:00, [(10:00, 11:00)]) → True

Source code in src/codex_services/booking/_shared/validators.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def is_slot_free(
    self,
    slot_start: datetime,
    slot_end: datetime,
    busy_intervals: list[tuple[datetime, datetime]],
) -> bool:
    """
    Verifies that the slot [slot_start, slot_end) does not overlap
    with any of the busy intervals.

    Uses a "half-open" interval — [start, end). If slot_end == busy_start,
    it is NOT considered a conflict (adjacent slots are allowed).

    Args:
        slot_start: Start of the slot to check.
        slot_end: End of the slot to check.
        busy_intervals: List of busy intervals [(start, end), ...].

    Returns:
        True if the slot is free. False if there is an overlap.

    Example:
        # Busy 10:00-11:00. Requesting 10:30-11:30 → conflict:
        is_slot_free(10:30, 11:30, [(10:00, 11:00)]) → False

        # Requesting 11:00-12:00 → OK (adjacent slots):
        is_slot_free(11:00, 12:00, [(10:00, 11:00)]) → True
    """
    return all(not (slot_start < busy_end and slot_end > busy_start) for busy_start, busy_end in busy_intervals)
no_conflicts(solutions)

Verifies that there are no resource conflicts within a set of solutions. A resource cannot be occupied by two services simultaneously.

Groups solutions by resource_id and checks each group for overlaps. Used by ChainFinder after assembling the chain for final verification.

Parameters:

Name Type Description Default
solutions list[BookingSolution]

List of BookingSolution objects (found slots).

 required

Returns:

Type Description
bool

True if no conflicts exist. False if at least one resource is double-booked.
Example

no_conflicts([ SingleServiceSolution(resource_id="1", start=9:00, gap_end=10:10), SingleServiceSolution(resource_id="1", start=10:10, gap_end=11:10), ])

→ True (slots are adjacent, no overlap)
Source code in src/codex_services/booking/_shared/validators.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def no_conflicts(
    self,
    solutions: list[BookingSolution],
) -> bool:
    """
    Verifies that there are no resource conflicts within a set of solutions.
    A resource cannot be occupied by two services simultaneously.

    Groups solutions by resource_id and checks each group for overlaps.
    Used by ChainFinder after assembling the chain for final verification.

    Args:
        solutions: List of BookingSolution objects (found slots).

    Returns:
        True if no conflicts exist. False if at least one resource is double-booked.

    Example:
        no_conflicts([
            SingleServiceSolution(resource_id="1", start=9:00, gap_end=10:10),
            SingleServiceSolution(resource_id="1", start=10:10, gap_end=11:10),
        ])
        # → True (slots are adjacent, no overlap)
    """
    by_resource: dict[str, list[BookingSolution]] = {}
    for sol in solutions:
        by_resource.setdefault(sol.resource_id, []).append(sol)

    for _resource_id, resource_solutions in by_resource.items():
        if len(resource_solutions) < 2:
            continue
        sorted_sols = sorted(resource_solutions, key=lambda s: s.start_time)
        for i in range(len(sorted_sols) - 1):
            current = sorted_sols[i]
            next_sol = sorted_sols[i + 1]
            # For solutions with gap_end_time, use it; otherwise use end_time
            block_end = getattr(current, "gap_end_time", current.end_time)
            if next_sol.start_time < block_end:
                return False

    return True
solution_fits_in_windows(solution, free_windows)

Verifies that a solution's slot fits entirely inside one of the resource's free windows.

Parameters:

Name Type Description Default
solution BookingSolution

Found slot for a single service.

 required
free_windows list[tuple[datetime, datetime]]

Resource's free windows (from ResourceAvailability).

 required

Returns:

Type Description
bool

True if the slot fits perfectly inside one of the given free windows.
Source code in src/codex_services/booking/_shared/validators.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def solution_fits_in_windows(
    self,
    solution: BookingSolution,
    free_windows: list[tuple[datetime, datetime]],
) -> bool:
    """
    Verifies that a solution's slot fits entirely inside one of the
    resource's free windows.

    Args:
        solution: Found slot for a single service.
        free_windows: Resource's free windows (from ResourceAvailability).

    Returns:
        True if the slot fits perfectly inside one of the given free windows.
    """
    block_end = getattr(solution, "gap_end_time", solution.end_time)
    return any(solution.start_time >= w_start and block_end <= w_end for w_start, w_end in free_windows)