Booking Internal Modules

The booking package is built from adapters, selectors, and abstract model mixins. This section indexes those implementation modules directly.

Selectors

codex_django.booking.selectors

codex_django.booking.selectors

Pure-function selectors for booking operations.

These are the high-level entry points for views and CLI-generated features. The adapter is passed as an argument (dependency injection).

Rules enforced

R1 — Cache invalidation via transaction.on_commit() only. R2 — lock_masters() uses select_for_update(of=('self',)).

Classes

BookingPersistenceHook

Bases: Protocol

Protocol for project-specific multi-service persistence.

Source code in src/codex_django/booking/selectors.py

class BookingPersistenceHook(Protocol):
    """Protocol for project-specific multi-service persistence."""

    def persist_chain(
        self,
        solution: Any,
        service_ids: list[int],
        client: Any,
        extra_fields: dict[str, Any] | None = None,
    ) -> list[Any]:
        """Persist a multi-service chain and return appointment-like objects."""
        ...

Functions

persist_chain(solution, service_ids, client, extra_fields=None)

Persist a multi-service chain and return appointment-like objects.

Source code in src/codex_django/booking/selectors.py

def persist_chain(
    self,
    solution: Any,
    service_ids: list[int],
    client: Any,
    extra_fields: dict[str, Any] | None = None,
) -> list[Any]:
    """Persist a multi-service chain and return appointment-like objects."""
    ...

Functions

get_available_slots(adapter, service_ids, target_date, *, locked_master_id=None, master_selections=None, mode=BookingMode.SINGLE_DAY, overlap_allowed=False, parallel_groups=None, max_solutions=50, max_unique_starts=None, cache_ttl=300)

Compute available booking slots for a given date.

Builds the engine request, fetches master availability, and runs ChainFinder.find().

Parameters:

Name	Type	Description	Default
adapter	DjangoAvailabilityAdapter	Availability adapter that bridges Django models to the engine.	required
service_ids	list[int]	Ordered list of requested service identifiers.	required
target_date	date	Date for which slots should be computed.	required
locked_master_id	int | None	Optional master id that constrains the search.	None
master_selections	MasterSelections	Optional per-service master selection mapping.	None
mode	BookingMode	Booking engine search mode.	SINGLE_DAY
overlap_allowed	bool	Whether services may overlap in time.	False
parallel_groups	dict[int, str] | None	Optional mapping of service id to parallel group id.	None
max_solutions	int	Maximum number of engine solutions to compute.	50
max_unique_starts	int | None	Optional cap for unique start times.	None
cache_ttl	int	Cache lifetime in seconds for busy-slot reads.	300

Returns:

Type	Description
EngineResult	Engine result produced by ChainFinder.find().

Source code in src/codex_django/booking/selectors.py

def get_available_slots(
    adapter: DjangoAvailabilityAdapter,
    service_ids: list[int],
    target_date: date,
    *,
    locked_master_id: int | None = None,
    master_selections: MasterSelections = None,
    mode: BookingMode = BookingMode.SINGLE_DAY,
    overlap_allowed: bool = False,
    parallel_groups: dict[int, str] | None = None,
    max_solutions: int = 50,
    max_unique_starts: int | None = None,
    cache_ttl: int = 300,
) -> EngineResult:
    """Compute available booking slots for a given date.

    Builds the engine request, fetches master availability, and runs
    ``ChainFinder.find()``.

    Args:
        adapter: Availability adapter that bridges Django models to the engine.
        service_ids: Ordered list of requested service identifiers.
        target_date: Date for which slots should be computed.
        locked_master_id: Optional master id that constrains the search.
        master_selections: Optional per-service master selection mapping.
        mode: Booking engine search mode.
        overlap_allowed: Whether services may overlap in time.
        parallel_groups: Optional mapping of service id to parallel group id.
        max_solutions: Maximum number of engine solutions to compute.
        max_unique_starts: Optional cap for unique start times.
        cache_ttl: Cache lifetime in seconds for busy-slot reads.

    Returns:
        Engine result produced by ``ChainFinder.find()``.
    """
    request = adapter.build_engine_request(
        service_ids=service_ids,
        target_date=target_date,
        locked_master_id=locked_master_id,
        master_selections=master_selections,
        mode=mode,
        overlap_allowed=overlap_allowed,
        parallel_groups=parallel_groups,
    )

    all_master_ids: list[int] = []
    for sr in request.service_requests:
        all_master_ids.extend(int(mid) for mid in sr.possible_resource_ids)
    unique_master_ids = list(set(all_master_ids))

    availability = adapter.build_masters_availability(
        master_ids=unique_master_ids,
        target_date=target_date,
        cache_ttl=cache_ttl,
    )

    finder = ChainFinder(step_minutes=adapter.step_minutes)
    return finder.find(
        request=request,
        resources_availability=availability,
        max_solutions=max_solutions,
        max_unique_starts=max_unique_starts,
    )

get_calendar_data(year, month, today=None, selected_date=None, holidays_subdiv='ST')

Generate a calendar month matrix for UI rendering.

Wraps CalendarEngine.get_month_matrix() from codex-services.

Parameters:

Name	Type	Description	Default
year	int	Target calendar year.	required
month	int	Target calendar month.	required
today	date | None	Optional date used to highlight the current day.	None
selected_date	date | None	Optional date selected in the UI.	None
holidays_subdiv	str	Region code passed to the calendar engine.	'ST'

Returns:

Type	Description
list[dict[str, Any]]	Calendar matrix data ready for template rendering.

Source code in src/codex_django/booking/selectors.py

def get_calendar_data(
    year: int,
    month: int,
    today: date | None = None,
    selected_date: date | None = None,
    holidays_subdiv: str = "ST",
) -> list[dict[str, Any]]:
    """Generate a calendar month matrix for UI rendering.

    Wraps ``CalendarEngine.get_month_matrix()`` from codex-services.

    Args:
        year: Target calendar year.
        month: Target calendar month.
        today: Optional date used to highlight the current day.
        selected_date: Optional date selected in the UI.
        holidays_subdiv: Region code passed to the calendar engine.

    Returns:
        Calendar matrix data ready for template rendering.
    """
    from codex_services.calendar.engine import CalendarEngine

    if today is None:
        today = date.today()

    return CalendarEngine.get_month_matrix(
        year=year,
        month=month,
        today=today,
        selected_date=selected_date,
        holidays_subdiv=holidays_subdiv,
    )

get_nearest_slots(adapter, service_ids, search_from, *, locked_master_id=None, master_selections=None, mode=BookingMode.SINGLE_DAY, overlap_allowed=False, parallel_groups=None, search_days=60, max_solutions_per_day=1)

Search for the nearest available date with open slots (waitlist).

Uses ChainFinder.find_nearest() which scans forward day by day.

Parameters:

Name	Type	Description	Default
adapter	DjangoAvailabilityAdapter	Availability adapter that bridges Django models to the engine.	required
service_ids	list[int]	Ordered list of requested service identifiers.	required
search_from	date	First date included in the forward search.	required
locked_master_id	int | None	Optional master id that constrains the search.	None
master_selections	MasterSelections	Optional per-service master selection mapping.	None
mode	BookingMode	Booking engine search mode.	SINGLE_DAY
overlap_allowed	bool	Whether services may overlap in time.	False
parallel_groups	dict[int, str] | None	Optional mapping of service id to parallel group id.	None
search_days	int	Number of forward days to inspect.	60
max_solutions_per_day	int	Maximum number of solutions evaluated per day.	1

Returns:

Type	Description
EngineResult	Engine result produced by ChainFinder.find_nearest().

Source code in src/codex_django/booking/selectors.py

def get_nearest_slots(
    adapter: DjangoAvailabilityAdapter,
    service_ids: list[int],
    search_from: date,
    *,
    locked_master_id: int | None = None,
    master_selections: MasterSelections = None,
    mode: BookingMode = BookingMode.SINGLE_DAY,
    overlap_allowed: bool = False,
    parallel_groups: dict[int, str] | None = None,
    search_days: int = 60,
    max_solutions_per_day: int = 1,
) -> EngineResult:
    """Search for the nearest available date with open slots (waitlist).

    Uses ``ChainFinder.find_nearest()`` which scans forward day by day.

    Args:
        adapter: Availability adapter that bridges Django models to the engine.
        service_ids: Ordered list of requested service identifiers.
        search_from: First date included in the forward search.
        locked_master_id: Optional master id that constrains the search.
        master_selections: Optional per-service master selection mapping.
        mode: Booking engine search mode.
        overlap_allowed: Whether services may overlap in time.
        parallel_groups: Optional mapping of service id to parallel group id.
        search_days: Number of forward days to inspect.
        max_solutions_per_day: Maximum number of solutions evaluated per day.

    Returns:
        Engine result produced by ``ChainFinder.find_nearest()``.
    """
    request = adapter.build_engine_request(
        service_ids=service_ids,
        target_date=search_from,
        locked_master_id=locked_master_id,
        master_selections=master_selections,
        mode=mode,
        overlap_allowed=overlap_allowed,
        parallel_groups=parallel_groups,
    )

    def get_availability_for_date(
        d: date,
    ) -> dict[str, Any]:
        all_master_ids: list[int] = []
        for sr in request.service_requests:
            all_master_ids.extend(int(mid) for mid in sr.possible_resource_ids)
        return adapter.build_masters_availability(
            master_ids=list(set(all_master_ids)),
            target_date=d,
        )

    finder = ChainFinder(step_minutes=adapter.step_minutes)
    return finder.find_nearest(
        request=request,
        get_availability_for_date=get_availability_for_date,
        search_from=search_from,
        search_days=search_days,
        max_solutions_per_day=max_solutions_per_day,
    )

create_booking(adapter, cache_adapter, appointment_model, *, service_ids, target_date, selected_time, master_id, client, extra_fields=None, master_selections=None, mode=BookingMode.SINGLE_DAY, overlap_allowed=False, parallel_groups=None, persistence_hook=None)

Create a booking with concurrency protection.

Solo mode: 1. Locks a single master row 2. Re-checks slot availability under lock 3. Creates one appointment 4. Invalidates cache after commit (R1: transaction.on_commit())

Multi-service mode: 1. Requires a persistence_hook 2. Locks all candidate masters for the chain 3. Re-checks chain availability under lock 4. Delegates persistence to persistence_hook.persist_chain() 5. Invalidates cache for chain masters after commit

Parameters:

Name	Type	Description	Default
adapter	DjangoAvailabilityAdapter	Availability adapter used for locking and revalidation.	required
cache_adapter	BookingCacheAdapter	Cache adapter used for post-commit invalidation.	required
appointment_model	type[Any]	Concrete Django model class for appointment rows.	required
service_ids	list[int]	Ordered list of requested service identifiers.	required
target_date	date	Booking date selected by the client.	required
selected_time	str	Selected start time in HH:MM format.	required
master_id	int	Master chosen for single-service mode.	required
client	Any	Client object attached to the booking.	required
extra_fields	dict[str, Any] | None	Optional extra model fields passed to persistence.	None
master_selections	MasterSelections	Optional per-service master selection mapping.	None
mode	BookingMode	Booking engine search mode.	SINGLE_DAY
overlap_allowed	bool	Whether services may overlap in time.	False
parallel_groups	dict[int, str] | None	Optional mapping of service id to parallel group id.	None
persistence_hook	BookingPersistenceHook | None	Required persistence hook for multi-service mode.	None

Returns:

Type	Description
Any | list[Any]	A single appointment instance in solo mode, or a list of created
Any | list[Any]	appointment-like objects in multi-service mode.

Raises:

Type	Description
SlotAlreadyBookedError	If the selected slot is no longer available under lock.
NotImplementedError	If multi-service mode is requested without a persistence hook.

Source code in src/codex_django/booking/selectors.py

def create_booking(
    adapter: DjangoAvailabilityAdapter,
    cache_adapter: BookingCacheAdapter,
    appointment_model: type[Any],
    *,
    service_ids: list[int],
    target_date: date,
    selected_time: str,
    master_id: int,
    client: Any,
    extra_fields: dict[str, Any] | None = None,
    master_selections: MasterSelections = None,
    mode: BookingMode = BookingMode.SINGLE_DAY,
    overlap_allowed: bool = False,
    parallel_groups: dict[int, str] | None = None,
    persistence_hook: BookingPersistenceHook | None = None,
) -> Any | list[Any]:
    """Create a booking with concurrency protection.

    Solo mode:
    1. Locks a single master row
    2. Re-checks slot availability under lock
    3. Creates one appointment
    4. Invalidates cache after commit (R1: ``transaction.on_commit()``)

    Multi-service mode:
    1. Requires a ``persistence_hook``
    2. Locks all candidate masters for the chain
    3. Re-checks chain availability under lock
    4. Delegates persistence to ``persistence_hook.persist_chain()``
    5. Invalidates cache for chain masters after commit

    Args:
        adapter: Availability adapter used for locking and revalidation.
        cache_adapter: Cache adapter used for post-commit invalidation.
        appointment_model: Concrete Django model class for appointment rows.
        service_ids: Ordered list of requested service identifiers.
        target_date: Booking date selected by the client.
        selected_time: Selected start time in ``HH:MM`` format.
        master_id: Master chosen for single-service mode.
        client: Client object attached to the booking.
        extra_fields: Optional extra model fields passed to persistence.
        master_selections: Optional per-service master selection mapping.
        mode: Booking engine search mode.
        overlap_allowed: Whether services may overlap in time.
        parallel_groups: Optional mapping of service id to parallel group id.
        persistence_hook: Required persistence hook for multi-service mode.

    Returns:
        A single appointment instance in solo mode, or a list of created
        appointment-like objects in multi-service mode.

    Raises:
        codex_services.booking.slot_master.SlotAlreadyBookedError: If the
            selected slot is no longer available under lock.
        NotImplementedError: If multi-service mode is requested without a
            persistence hook.
    """
    is_multi_service = len(service_ids) > 1

    with transaction.atomic():
        if is_multi_service:
            if persistence_hook is None:
                raise NotImplementedError("Multi-service persistence requires a persistence hook")

            request = adapter.build_engine_request(
                service_ids=service_ids,
                target_date=target_date,
                locked_master_id=None,
                master_selections=master_selections,
                mode=mode,
                overlap_allowed=overlap_allowed,
                parallel_groups=parallel_groups,
            )
            all_master_ids: set[int] = {int(mid) for sr in request.service_requests for mid in sr.possible_resource_ids}
            adapter.lock_masters(sorted(all_master_ids))

            result = get_available_slots(
                adapter=adapter,
                service_ids=service_ids,
                target_date=target_date,
                locked_master_id=None,
                master_selections=master_selections,
                mode=mode,
                overlap_allowed=overlap_allowed,
                parallel_groups=parallel_groups,
                cache_ttl=0,  # no cache under lock — fresh data
            )
            available_times = result.get_unique_start_times()
            if selected_time not in available_times:
                raise SlotAlreadyBookedError(f"Slot {selected_time} on {target_date} is no longer available.")

            best = result.best
            if best is None:
                raise SlotAlreadyBookedError("No solution found.")

            created_appointments = persistence_hook.persist_chain(
                solution=best,
                service_ids=service_ids,
                client=client,
                extra_fields=extra_fields,
            )

            chain_master_ids = {
                str(item.resource_id) for item in getattr(best, "items", []) if getattr(item, "resource_id", None)
            }
            if not chain_master_ids:
                chain_master_ids = {str(master_id)}

            _date_str = str(target_date)

            def _invalidate_many() -> None:
                for mid in chain_master_ids:
                    cache_adapter.invalidate_master_date(mid, _date_str)

            transaction.on_commit(_invalidate_many)
            return created_appointments

        adapter.lock_masters([master_id])

        result = get_available_slots(
            adapter=adapter,
            service_ids=service_ids,
            target_date=target_date,
            locked_master_id=master_id,
            master_selections=master_selections,
            mode=mode,
            overlap_allowed=overlap_allowed,
            parallel_groups=parallel_groups,
            cache_ttl=0,  # no cache under lock — fresh data
        )

        available_times = result.get_unique_start_times()
        if selected_time not in available_times:
            raise SlotAlreadyBookedError(f"Slot {selected_time} on {target_date} is no longer available.")

        # Find the matching solution for duration info
        best = result.best
        if best is None:
            raise SlotAlreadyBookedError("No solution found.")

        fields: dict[str, Any] = {
            "master_id": master_id,
            "datetime_start": best.starts_at,
            "duration_minutes": best.span_minutes,
            "client": client,
        }
        if extra_fields:
            fields.update(extra_fields)

        appointment = appointment_model.objects.create(**fields)

        # R1: invalidate cache AFTER transaction commits
        _master_id_str = str(master_id)
        _date_str = str(target_date)
        transaction.on_commit(lambda: cache_adapter.invalidate_master_date(_master_id_str, _date_str))

    return appointment

Availability adapter

codex_django.booking.adapters.availability

codex_django.booking.adapters.availability

Universal bridge between Django ORM and the codex-services booking engine.

All models are injected at construction time — no hardcoded imports.

Rules enforced

R1 — Cache invalidation via transaction.on_commit() only (see selectors). R2 — select_for_update(of=('self',)) in lock_masters(). R3 — No clean/save overrides here; pure adapter logic.

Classes

DjangoAvailabilityAdapter

Universal bridge between Django ORM and the booking engine.

Implements the three provider interfaces from codex-services so it can be used directly with ChainFinder.

Parameters:

Name	Type	Description	Default
master_model	type[Any]	Django model class for masters/resources.	required
appointment_model	type[Any]	Django model class for appointments.	required
service_model	type[Any]	Django model class for services.	required
working_day_model	type[Any]	Django model class for per-weekday schedule.	required
day_off_model	type[Any]	Django model class for days off.	required
booking_settings_model	type[Any]	Django model class for booking settings.	required
site_settings_model	type[Any]	Django model class for site-level settings.	required
step_minutes	int	Time grid step for the engine.	30
appointment_status_filter	list[str] | None	Which statuses count as "busy".	None
cache_adapter	BookingCacheAdapter | None	Optional cache adapter; defaults to BookingCacheAdapter.	None

Source code in src/codex_django/booking/adapters/availability.py

class DjangoAvailabilityAdapter:
    """Universal bridge between Django ORM and the booking engine.

    Implements the three provider interfaces from codex-services so it
    can be used directly with ``ChainFinder``.

    Args:
        master_model: Django model class for masters/resources.
        appointment_model: Django model class for appointments.
        service_model: Django model class for services.
        working_day_model: Django model class for per-weekday schedule.
        day_off_model: Django model class for days off.
        booking_settings_model: Django model class for booking settings.
        site_settings_model: Django model class for site-level settings.
        step_minutes: Time grid step for the engine.
        appointment_status_filter: Which statuses count as "busy".
        cache_adapter: Optional cache adapter; defaults to BookingCacheAdapter.
    """

    def __init__(
        self,
        master_model: type[Any],
        appointment_model: type[Any],
        service_model: type[Any],
        working_day_model: type[Any],
        day_off_model: type[Any],
        booking_settings_model: type[Any],
        site_settings_model: type[Any],
        step_minutes: int = 30,
        appointment_status_filter: list[str] | None = None,
        cache_adapter: BookingCacheAdapter | None = None,
    ) -> None:
        self.master_model = master_model
        self.appointment_model = appointment_model
        self.service_model = service_model
        self.working_day_model = working_day_model
        self.day_off_model = day_off_model
        self.booking_settings_model = booking_settings_model
        self.site_settings_model = site_settings_model

        self.step_minutes = step_minutes
        self._calc = SlotCalculator(step_minutes)
        self._cache = cache_adapter or BookingCacheAdapter()

        if appointment_status_filter:
            self.appointment_status_filter = appointment_status_filter
        else:
            self.appointment_status_filter = [
                getattr(appointment_model, "STATUS_PENDING", "pending"),
                getattr(appointment_model, "STATUS_CONFIRMED", "confirmed"),
            ]
            if hasattr(appointment_model, "STATUS_RESCHEDULE_PROPOSED"):
                self.appointment_status_filter.append(appointment_model.STATUS_RESCHEDULE_PROPOSED)

        self._booking_settings: Any = None
        self._site_settings: Any = None

    # ------------------------------------------------------------------
    # Engine request builder
    # ------------------------------------------------------------------

    def build_engine_request(
        self,
        service_ids: list[int],
        target_date: date,
        locked_master_id: int | None = None,
        master_selections: dict[str, str] | dict[int, int | None] | None = None,
        mode: BookingMode = BookingMode.SINGLE_DAY,
        overlap_allowed: bool = False,
        parallel_groups: dict[int, str] | None = None,
    ) -> BookingEngineRequest:
        """Build a ``BookingEngineRequest`` from DB service/master data."""
        services = self.service_model.objects.filter(id__in=service_ids).select_related("category")
        service_map = {s.id: s for s in services}
        normalized_master_selections = self._normalize_master_selections(service_ids, master_selections)

        weekday = target_date.weekday()
        service_requests: list[ServiceRequest] = []

        for svc_id in service_ids:
            service = service_map.get(svc_id)
            if not service:
                continue

            possible_ids = self._resolve_master_ids(
                service=service,
                weekday=weekday,
                locked_master_id=locked_master_id,
                master_selections=normalized_master_selections,
                service_id=svc_id,
            )
            if not possible_ids:
                continue

            gap = getattr(service, "min_gap_after_minutes", 0) or 0
            parallel_group = (parallel_groups or {}).get(svc_id)
            if parallel_group is None:
                parallel_group = getattr(service, "parallel_group", None) or None

            service_requests.append(
                ServiceRequest(
                    service_id=str(svc_id),
                    duration_minutes=service.duration,
                    min_gap_after_minutes=gap,
                    possible_resource_ids=possible_ids,
                    parallel_group=parallel_group,
                )
            )

        return BookingEngineRequest(
            service_requests=service_requests,
            booking_date=target_date,
            mode=mode,
            overlap_allowed=overlap_allowed,
        )

    # ------------------------------------------------------------------
    # Availability builder (implements AvailabilityProvider pattern)
    # ------------------------------------------------------------------

    def build_masters_availability(
        self,
        master_ids: list[int],
        target_date: date,
        cache_ttl: int = 0,
        exclude_appointment_ids: list[int] | None = None,
    ) -> dict[str, MasterAvailability]:
        """Build ``MasterAvailability`` dicts from ORM data.

        Caches **busy intervals** (not free slots) per master+date.
        """
        settings = self._get_booking_settings()
        result: dict[str, MasterAvailability] = {}

        # Days off — exclude these masters entirely
        day_off_ids = set(
            self.day_off_model.objects.filter(master_id__in=master_ids, date=target_date).values_list(
                "master_id", flat=True
            )
        )

        # Busy intervals (from appointments)
        busy_by_master = self._get_busy_intervals(master_ids, target_date, exclude_appointment_ids)

        masters = self.master_model.objects.filter(pk__in=master_ids)
        for master in masters:
            if master.pk in day_off_ids:
                continue

            working_hours_utc = self.get_working_hours(master, target_date)
            if not working_hours_utc:
                continue

            work_start_dt, work_end_dt = working_hours_utc
            break_interval = self.get_break_interval(master, target_date)
            buffer = self._get_buffer_minutes(master, settings)

            free_windows = self._calc.merge_free_windows(
                work_start=work_start_dt,
                work_end=work_end_dt,
                busy_intervals=busy_by_master.get(master.pk, []),
                break_interval=break_interval,
                buffer_minutes=buffer,
                min_duration_minutes=self.step_minutes,
            )

            result[str(master.pk)] = MasterAvailability(
                resource_id=str(master.pk),
                free_windows=free_windows,
                buffer_between_minutes=buffer,
            )

        return result

    # ------------------------------------------------------------------
    # Schedule (implements ScheduleProvider pattern)
    # ------------------------------------------------------------------

    def get_working_hours(self, master: Any, target_date: date) -> tuple[datetime, datetime] | None:
        """Return UTC working hours for a master on a specific date.

        Reads from ``working_day_model`` first; falls back to
        master-level defaults, then site-level defaults.
        """
        weekday = target_date.weekday()
        tz = self._get_tz(master)

        # 1. Per-day schedule from relational model
        working_day = self.working_day_model.objects.filter(master_id=master.pk, weekday=weekday).first()

        if working_day:
            start_t = working_day.start_time
            end_t = working_day.end_time
        else:
            # 2. Master-level defaults
            start_t = getattr(master, "work_start", None)
            end_t = getattr(master, "work_end", None)

        if not (start_t and end_t):
            # 3. Site-level defaults
            site = self._get_site_settings()
            if weekday < 5:
                start_t = getattr(site, "work_start_weekdays", None)
                end_t = getattr(site, "work_end_weekdays", None)
            elif weekday == 5:
                start_t = getattr(site, "work_start_saturday", None)
                end_t = getattr(site, "work_end_saturday", None)

        if not start_t or not end_t:
            return None

        work_start_dt = datetime.combine(target_date, start_t, tzinfo=tz)
        work_end_dt = datetime.combine(target_date, end_t, tzinfo=tz)
        return (work_start_dt.astimezone(UTC), work_end_dt.astimezone(UTC))

    def get_break_interval(self, master: Any, target_date: date) -> tuple[datetime, datetime] | None:
        """Return UTC break interval for a master on a date."""
        weekday = target_date.weekday()

        # Per-day schedule first
        working_day = self.working_day_model.objects.filter(master_id=master.pk, weekday=weekday).first()

        if working_day:
            break_start = working_day.break_start
            break_end = working_day.break_end
        else:
            break_start = getattr(master, "break_start", None)
            break_end = getattr(master, "break_end", None)

        if not (break_start and break_end):
            return None

        tz = self._get_tz(master)
        bs = datetime.combine(target_date, break_start, tzinfo=tz)
        be = datetime.combine(target_date, break_end, tzinfo=tz)
        return (bs.astimezone(UTC), be.astimezone(UTC))

    # ------------------------------------------------------------------
    # Locking (R2: select_for_update with of=('self',))
    # ------------------------------------------------------------------

    def lock_masters(self, master_ids: list[int]) -> None:
        """Acquire row-level locks on master records.

        Must be called inside ``transaction.atomic()``.
        IDs are sorted to prevent deadlocks.
        """
        if not master_ids:
            return
        sorted_ids = sorted(master_ids)
        list(self.master_model.objects.select_for_update(of=("self",)).filter(pk__in=sorted_ids).only("pk"))

    # ------------------------------------------------------------------
    # Busy intervals (implements BusySlotsProvider pattern)
    # ------------------------------------------------------------------

    def _get_busy_intervals(
        self,
        master_ids: list[int],
        target_date: date,
        exclude_appointment_ids: list[int] | None = None,
    ) -> dict[int, list[tuple[datetime, datetime]]]:
        """Fetch busy intervals from cache or DB."""
        busy_by_master: dict[int, list[tuple[datetime, datetime]]] = {mid: [] for mid in master_ids}

        appt_filter = {
            "master_id__in": master_ids,
            "datetime_start__date": target_date,
            "status__in": self.appointment_status_filter,
        }
        appointments_qs = self.appointment_model.objects.filter(**appt_filter)
        if exclude_appointment_ids:
            appointments_qs = appointments_qs.exclude(id__in=exclude_appointment_ids)
        appointments = appointments_qs.order_by("datetime_start")

        for app in appointments:
            s = app.datetime_start.astimezone(UTC).replace(second=0, microsecond=0)
            e = s + timedelta(minutes=app.duration_minutes)
            busy_by_master[app.master_id].append((s, e))

        return busy_by_master

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def result_to_slots_map(self, result: EngineResult) -> dict[str, bool]:
        """Convert an engine result to a simple ``{HH:MM: True}`` map."""
        times = result.get_unique_start_times()
        return dict.fromkeys(times, True)

    def _resolve_master_ids(
        self,
        service: Any,
        weekday: int,
        locked_master_id: int | None,
        master_selections: dict[int, int | None] | None,
        service_id: int,
    ) -> list[str]:
        """Determine which master IDs can perform a service."""
        if locked_master_id:
            return [str(locked_master_id)]

        if master_selections and service_id in master_selections:
            m_id = master_selections[service_id]
            if m_id is not None:
                return [str(m_id)]

        # All active masters for this service's category working on this weekday
        masters = self.master_model.objects.filter(
            categories=service.category,
            status=self.master_model.STATUS_ACTIVE,
        )

        # Filter by weekday using the relational working_day_model
        masters_with_schedule = set(
            self.working_day_model.objects.filter(
                master_id__in=masters.values_list("pk", flat=True),
                weekday=weekday,
            ).values_list("master_id", flat=True)
        )

        return [str(m.pk) for m in masters if m.pk in masters_with_schedule]

    def _normalize_master_selections(
        self,
        service_ids: list[int],
        master_selections: dict[str, str] | dict[int, int | None] | None,
    ) -> dict[int, int | None]:
        """Normalize legacy and new master selection formats.

        Supported input formats:
        - Legacy positional: {"0": "10", "1": "12"}
        - Service keyed: {5: 10, 7: None} or {"5": "10", "7": "12"}
        """
        if not master_selections:
            return {}

        # Legacy positional format: keys are indices inside service_ids.
        is_legacy_positional = all(
            isinstance(key, str) and key.isdigit() and int(key) < len(service_ids) for key in master_selections
        )
        if is_legacy_positional:
            normalized_legacy: dict[int, int | None] = {}
            for key, raw_val in master_selections.items():
                idx = int(key)
                if raw_val in (None, "any", ""):
                    continue
                if not isinstance(raw_val, str | int):
                    continue
                normalized_legacy[service_ids[idx]] = int(raw_val)
            return normalized_legacy

        normalized: dict[int, int | None] = {}
        for raw_key, raw_val in master_selections.items():
            service_id = int(raw_key)
            if raw_val in (None, "any", ""):
                normalized[service_id] = None
            else:
                if not isinstance(raw_val, str | int):
                    continue
                normalized[service_id] = int(raw_val)
        return normalized

    def _get_buffer_minutes(self, master: Any, settings: Any) -> int:
        """Return the effective buffer duration for a master."""
        individual = getattr(master, "buffer_between_minutes", None)
        if individual is not None:
            return individual  # type: ignore[no-any-return]
        return getattr(settings, "default_buffer_between_minutes", 0)

    def _get_tz(self, master: Any) -> zoneinfo.ZoneInfo:
        """Resolve the master timezone, falling back to the site timezone."""
        tz_name = getattr(master, "timezone", None)
        if not tz_name:
            site = self._get_site_settings()
            tz_name = getattr(site, "timezone", None) or "UTC"
        try:
            return zoneinfo.ZoneInfo(tz_name)
        except Exception:
            return zoneinfo.ZoneInfo("UTC")

    def _get_booking_settings(self) -> Any:
        """Lazy-load and memoize the booking settings singleton."""
        if self._booking_settings is None:
            self._booking_settings = self.booking_settings_model.objects.first()
        return self._booking_settings

    def _get_site_settings(self) -> Any:
        """Lazy-load and memoize the site settings singleton."""
        if self._site_settings is None:
            self._site_settings = self.site_settings_model.objects.first()
        return self._site_settings

Functions

build_engine_request(service_ids, target_date, locked_master_id=None, master_selections=None, mode=BookingMode.SINGLE_DAY, overlap_allowed=False, parallel_groups=None)

Build a BookingEngineRequest from DB service/master data.

Source code in src/codex_django/booking/adapters/availability.py

def build_engine_request(
    self,
    service_ids: list[int],
    target_date: date,
    locked_master_id: int | None = None,
    master_selections: dict[str, str] | dict[int, int | None] | None = None,
    mode: BookingMode = BookingMode.SINGLE_DAY,
    overlap_allowed: bool = False,
    parallel_groups: dict[int, str] | None = None,
) -> BookingEngineRequest:
    """Build a ``BookingEngineRequest`` from DB service/master data."""
    services = self.service_model.objects.filter(id__in=service_ids).select_related("category")
    service_map = {s.id: s for s in services}
    normalized_master_selections = self._normalize_master_selections(service_ids, master_selections)

    weekday = target_date.weekday()
    service_requests: list[ServiceRequest] = []

    for svc_id in service_ids:
        service = service_map.get(svc_id)
        if not service:
            continue

        possible_ids = self._resolve_master_ids(
            service=service,
            weekday=weekday,
            locked_master_id=locked_master_id,
            master_selections=normalized_master_selections,
            service_id=svc_id,
        )
        if not possible_ids:
            continue

        gap = getattr(service, "min_gap_after_minutes", 0) or 0
        parallel_group = (parallel_groups or {}).get(svc_id)
        if parallel_group is None:
            parallel_group = getattr(service, "parallel_group", None) or None

        service_requests.append(
            ServiceRequest(
                service_id=str(svc_id),
                duration_minutes=service.duration,
                min_gap_after_minutes=gap,
                possible_resource_ids=possible_ids,
                parallel_group=parallel_group,
            )
        )

    return BookingEngineRequest(
        service_requests=service_requests,
        booking_date=target_date,
        mode=mode,
        overlap_allowed=overlap_allowed,
    )

build_masters_availability(master_ids, target_date, cache_ttl=0, exclude_appointment_ids=None)

Build MasterAvailability dicts from ORM data.

Caches busy intervals (not free slots) per master+date.

Source code in src/codex_django/booking/adapters/availability.py

def build_masters_availability(
    self,
    master_ids: list[int],
    target_date: date,
    cache_ttl: int = 0,
    exclude_appointment_ids: list[int] | None = None,
) -> dict[str, MasterAvailability]:
    """Build ``MasterAvailability`` dicts from ORM data.

    Caches **busy intervals** (not free slots) per master+date.
    """
    settings = self._get_booking_settings()
    result: dict[str, MasterAvailability] = {}

    # Days off — exclude these masters entirely
    day_off_ids = set(
        self.day_off_model.objects.filter(master_id__in=master_ids, date=target_date).values_list(
            "master_id", flat=True
        )
    )

    # Busy intervals (from appointments)
    busy_by_master = self._get_busy_intervals(master_ids, target_date, exclude_appointment_ids)

    masters = self.master_model.objects.filter(pk__in=master_ids)
    for master in masters:
        if master.pk in day_off_ids:
            continue

        working_hours_utc = self.get_working_hours(master, target_date)
        if not working_hours_utc:
            continue

        work_start_dt, work_end_dt = working_hours_utc
        break_interval = self.get_break_interval(master, target_date)
        buffer = self._get_buffer_minutes(master, settings)

        free_windows = self._calc.merge_free_windows(
            work_start=work_start_dt,
            work_end=work_end_dt,
            busy_intervals=busy_by_master.get(master.pk, []),
            break_interval=break_interval,
            buffer_minutes=buffer,
            min_duration_minutes=self.step_minutes,
        )

        result[str(master.pk)] = MasterAvailability(
            resource_id=str(master.pk),
            free_windows=free_windows,
            buffer_between_minutes=buffer,
        )

    return result

get_working_hours(master, target_date)

Return UTC working hours for a master on a specific date.

Reads from working_day_model first; falls back to master-level defaults, then site-level defaults.

Source code in src/codex_django/booking/adapters/availability.py

def get_working_hours(self, master: Any, target_date: date) -> tuple[datetime, datetime] | None:
    """Return UTC working hours for a master on a specific date.

    Reads from ``working_day_model`` first; falls back to
    master-level defaults, then site-level defaults.
    """
    weekday = target_date.weekday()
    tz = self._get_tz(master)

    # 1. Per-day schedule from relational model
    working_day = self.working_day_model.objects.filter(master_id=master.pk, weekday=weekday).first()

    if working_day:
        start_t = working_day.start_time
        end_t = working_day.end_time
    else:
        # 2. Master-level defaults
        start_t = getattr(master, "work_start", None)
        end_t = getattr(master, "work_end", None)

    if not (start_t and end_t):
        # 3. Site-level defaults
        site = self._get_site_settings()
        if weekday < 5:
            start_t = getattr(site, "work_start_weekdays", None)
            end_t = getattr(site, "work_end_weekdays", None)
        elif weekday == 5:
            start_t = getattr(site, "work_start_saturday", None)
            end_t = getattr(site, "work_end_saturday", None)

    if not start_t or not end_t:
        return None

    work_start_dt = datetime.combine(target_date, start_t, tzinfo=tz)
    work_end_dt = datetime.combine(target_date, end_t, tzinfo=tz)
    return (work_start_dt.astimezone(UTC), work_end_dt.astimezone(UTC))

get_break_interval(master, target_date)

Return UTC break interval for a master on a date.

Source code in src/codex_django/booking/adapters/availability.py

def get_break_interval(self, master: Any, target_date: date) -> tuple[datetime, datetime] | None:
    """Return UTC break interval for a master on a date."""
    weekday = target_date.weekday()

    # Per-day schedule first
    working_day = self.working_day_model.objects.filter(master_id=master.pk, weekday=weekday).first()

    if working_day:
        break_start = working_day.break_start
        break_end = working_day.break_end
    else:
        break_start = getattr(master, "break_start", None)
        break_end = getattr(master, "break_end", None)

    if not (break_start and break_end):
        return None

    tz = self._get_tz(master)
    bs = datetime.combine(target_date, break_start, tzinfo=tz)
    be = datetime.combine(target_date, break_end, tzinfo=tz)
    return (bs.astimezone(UTC), be.astimezone(UTC))

lock_masters(master_ids)

Acquire row-level locks on master records.

Must be called inside transaction.atomic(). IDs are sorted to prevent deadlocks.

Source code in src/codex_django/booking/adapters/availability.py

def lock_masters(self, master_ids: list[int]) -> None:
    """Acquire row-level locks on master records.

    Must be called inside ``transaction.atomic()``.
    IDs are sorted to prevent deadlocks.
    """
    if not master_ids:
        return
    sorted_ids = sorted(master_ids)
    list(self.master_model.objects.select_for_update(of=("self",)).filter(pk__in=sorted_ids).only("pk"))

result_to_slots_map(result)

Convert an engine result to a simple {HH:MM: True} map.

Source code in src/codex_django/booking/adapters/availability.py

def result_to_slots_map(self, result: EngineResult) -> dict[str, bool]:
    """Convert an engine result to a simple ``{HH:MM: True}`` map."""
    times = result.get_unique_start_times()
    return dict.fromkeys(times, True)

Cache adapter

codex_django.booking.adapters.cache

codex_django.booking.adapters.cache

Thin adapter over BookingCacheManager for busy slot caching.

Follows the same pattern as notifications.adapters.cache_adapter.DjangoCacheAdapter.

Classes

BookingCacheAdapter

Delegates to BookingCacheManager.

Used by DjangoAvailabilityAdapter to cache/invalidate busy intervals.

Source code in src/codex_django/booking/adapters/cache.py

class BookingCacheAdapter:
    """Delegates to BookingCacheManager.

    Used by DjangoAvailabilityAdapter to cache/invalidate busy intervals.
    """

    def get_busy(self, master_id: str, date_str: str) -> list[list[str]] | None:
        """Return cached busy intervals for a master on a specific date."""
        manager = get_booking_cache_manager()
        return manager.get_busy(master_id, date_str)

    def set_busy(
        self,
        master_id: str,
        date_str: str,
        intervals: list[list[str]],
        timeout: int = 300,
    ) -> None:
        """Store busy intervals for a master/date cache bucket."""
        manager = get_booking_cache_manager()
        manager.set_busy(master_id, date_str, intervals, timeout=timeout)

    def invalidate_master_date(self, master_id: str, date_str: str) -> None:
        """Invalidate the busy-interval cache for one master/date pair."""
        manager = get_booking_cache_manager()
        manager.invalidate_master_date(master_id, date_str)

Functions

get_busy(master_id, date_str)

Return cached busy intervals for a master on a specific date.

Source code in src/codex_django/booking/adapters/cache.py

def get_busy(self, master_id: str, date_str: str) -> list[list[str]] | None:
    """Return cached busy intervals for a master on a specific date."""
    manager = get_booking_cache_manager()
    return manager.get_busy(master_id, date_str)

set_busy(master_id, date_str, intervals, timeout=300)

Store busy intervals for a master/date cache bucket.

Source code in src/codex_django/booking/adapters/cache.py

def set_busy(
    self,
    master_id: str,
    date_str: str,
    intervals: list[list[str]],
    timeout: int = 300,
) -> None:
    """Store busy intervals for a master/date cache bucket."""
    manager = get_booking_cache_manager()
    manager.set_busy(master_id, date_str, intervals, timeout=timeout)

invalidate_master_date(master_id, date_str)

Invalidate the busy-interval cache for one master/date pair.

Source code in src/codex_django/booking/adapters/cache.py

def invalidate_master_date(self, master_id: str, date_str: str) -> None:
    """Invalidate the busy-interval cache for one master/date pair."""
    manager = get_booking_cache_manager()
    manager.invalidate_master_date(master_id, date_str)

Functions

Service mixins

codex_django.booking.mixins.service

codex_django.booking.mixins.service

Composable mixins for the bookable service model.

Usage::

from codex_django.booking.mixins import AbstractBookableService

class Service(AbstractBookableService):
    name = models.CharField(max_length=255)
    price = models.DecimalField(...)
    category = models.ForeignKey(...)

    class Meta:
        verbose_name = _("Service")

Classes

ServiceDurationMixin

Bases: Model

Service duration in minutes.

Admin fieldsets example::

(_("Duration"), {
    "fields": ("duration",),
}),

Source code in src/codex_django/booking/mixins/service.py

class ServiceDurationMixin(models.Model):
    """Service duration in minutes.

    Admin fieldsets example::

        (_("Duration"), {
            "fields": ("duration",),
        }),
    """

    duration = models.PositiveIntegerField(
        _("Duration (min)"),
        help_text=_("Service duration in minutes."),
    )

    class Meta:
        abstract = True

ServiceGapMixin

Bases: Model

Minimum gap after this service before the next one starts.

Admin fieldsets example::

(_("Gap"), {
    "fields": ("min_gap_after_minutes",),
    "classes": ("collapse",),
}),

Source code in src/codex_django/booking/mixins/service.py

class ServiceGapMixin(models.Model):
    """Minimum gap after this service before the next one starts.

    Admin fieldsets example::

        (_("Gap"), {
            "fields": ("min_gap_after_minutes",),
            "classes": ("collapse",),
        }),
    """

    min_gap_after_minutes = models.PositiveIntegerField(
        _("Gap After Service (min)"),
        default=0,
        help_text=_("Cleanup/preparation time after this service."),
    )

    class Meta:
        abstract = True

ServiceParallelMixin

Bases: Model

Controls whether this service can run in parallel with others.

Admin fieldsets example::

(_("Parallel Booking"), {
    "fields": ("parallel_ok", "parallel_group"),
    "classes": ("collapse",),
}),

Source code in src/codex_django/booking/mixins/service.py

class ServiceParallelMixin(models.Model):
    """Controls whether this service can run in parallel with others.

    Admin fieldsets example::

        (_("Parallel Booking"), {
            "fields": ("parallel_ok", "parallel_group"),
            "classes": ("collapse",),
        }),
    """

    parallel_ok = models.BooleanField(
        _("Allow Parallel"),
        default=True,
        help_text=_("Whether this service can be performed alongside another."),
    )
    parallel_group = models.CharField(
        _("Parallel Group"),
        max_length=50,
        blank=True,
        help_text=_("Services in the same group can run simultaneously."),
    )

    class Meta:
        abstract = True

AbstractBookableService

Bases: ServiceDurationMixin, ServiceGapMixin, ServiceParallelMixin, Model

Convenience base that assembles all service mixins.

Usage::

class Service(AbstractBookableService):
    name = models.CharField(max_length=255)
    price = models.DecimalField(max_digits=10, decimal_places=2)

    class Meta:
        verbose_name = _("Service")

Source code in src/codex_django/booking/mixins/service.py

class AbstractBookableService(
    ServiceDurationMixin,
    ServiceGapMixin,
    ServiceParallelMixin,
    models.Model,
):
    """Convenience base that assembles all service mixins.

    Usage::

        class Service(AbstractBookableService):
            name = models.CharField(max_length=255)
            price = models.DecimalField(max_digits=10, decimal_places=2)

            class Meta:
                verbose_name = _("Service")
    """

    class Meta:
        abstract = True

Master mixins

codex_django.booking.mixins.master

codex_django.booking.mixins.master

Composable mixins for the master/resource/specialist model.

Usage::

from codex_django.booking.mixins import AbstractBookableMaster

class Master(AbstractBookableMaster):
    name = models.CharField(max_length=255)
    # ... project-specific fields

    class Meta:
        verbose_name = _("Master")

Or pick individual mixins::

class Master(MasterScheduleMixin, MasterBufferMixin, models.Model):
    ...

Classes

MasterScheduleMixin

Bases: Model

Default working hours and breaks.

These serve as fallback values. Per-day overrides live in the AbstractWorkingDay relational model (see schedule.py).

Admin fieldsets example::

(_("Working Hours"), {
    "fields": (
        "work_start", "work_end",
        "break_start", "break_end",
    ),
}),

Source code in src/codex_django/booking/mixins/master.py

class MasterScheduleMixin(models.Model):
    """Default working hours and breaks.

    These serve as fallback values. Per-day overrides live in the
    ``AbstractWorkingDay`` relational model (see ``schedule.py``).

    Admin fieldsets example::

        (_("Working Hours"), {
            "fields": (
                "work_start", "work_end",
                "break_start", "break_end",
            ),
        }),
    """

    work_start = models.TimeField(_("Work Start"), null=True, blank=True)
    work_end = models.TimeField(_("Work End"), null=True, blank=True)
    break_start = models.TimeField(_("Break Start"), null=True, blank=True)
    break_end = models.TimeField(_("Break End"), null=True, blank=True)

    class Meta:
        abstract = True

MasterBufferMixin

Bases: Model

Buffer time and advance booking limits.

Admin fieldsets example::

(_("Booking Constraints"), {
    "fields": (
        "buffer_between_minutes",
        "min_advance_minutes",
        "max_advance_days",
    ),
    "classes": ("collapse",),
}),

Source code in src/codex_django/booking/mixins/master.py

class MasterBufferMixin(models.Model):
    """Buffer time and advance booking limits.

    Admin fieldsets example::

        (_("Booking Constraints"), {
            "fields": (
                "buffer_between_minutes",
                "min_advance_minutes",
                "max_advance_days",
            ),
            "classes": ("collapse",),
        }),
    """

    buffer_between_minutes = models.PositiveIntegerField(
        _("Buffer Between Appointments (min)"),
        null=True,
        blank=True,
        help_text=_("Overrides global default when set."),
    )
    min_advance_minutes = models.PositiveIntegerField(
        _("Minimum Advance Booking (min)"),
        null=True,
        blank=True,
        help_text=_("How far in advance a client must book."),
    )
    max_advance_days = models.PositiveIntegerField(
        _("Maximum Advance Booking (days)"),
        null=True,
        blank=True,
        help_text=_("How far ahead a client can book."),
    )

    class Meta:
        abstract = True

MasterCapacityMixin

Bases: Model

Parallel client capacity.

Admin fieldsets example::

(_("Capacity"), {
    "fields": ("max_clients_parallel",),
}),

Source code in src/codex_django/booking/mixins/master.py

class MasterCapacityMixin(models.Model):
    """Parallel client capacity.

    Admin fieldsets example::

        (_("Capacity"), {
            "fields": ("max_clients_parallel",),
        }),
    """

    max_clients_parallel = models.PositiveSmallIntegerField(
        _("Max Parallel Clients"),
        default=1,
    )

    class Meta:
        abstract = True

MasterTimezoneMixin

Bases: Model

Individual timezone for the master.

Falls back to site-level timezone when empty.

Admin fieldsets example::

(_("Timezone"), {
    "fields": ("timezone",),
    "classes": ("collapse",),
}),

Source code in src/codex_django/booking/mixins/master.py

class MasterTimezoneMixin(models.Model):
    """Individual timezone for the master.

    Falls back to site-level timezone when empty.

    Admin fieldsets example::

        (_("Timezone"), {
            "fields": ("timezone",),
            "classes": ("collapse",),
        }),
    """

    timezone = models.CharField(
        _("Timezone"),
        max_length=64,
        default="UTC",
        blank=True,
    )

    class Meta:
        abstract = True

AbstractBookableMaster

Bases: MasterScheduleMixin, MasterBufferMixin, MasterCapacityMixin, MasterTimezoneMixin, Model

Convenience base that assembles all master mixins.

Usage::

class Master(AbstractBookableMaster):
    name = models.CharField(max_length=255)
    status = models.CharField(...)
    categories = models.ManyToManyField(...)

    class Meta:
        verbose_name = _("Master")

Source code in src/codex_django/booking/mixins/master.py

class AbstractBookableMaster(
    MasterScheduleMixin,
    MasterBufferMixin,
    MasterCapacityMixin,
    MasterTimezoneMixin,
    models.Model,
):
    """Convenience base that assembles all master mixins.

    Usage::

        class Master(AbstractBookableMaster):
            name = models.CharField(max_length=255)
            status = models.CharField(...)
            categories = models.ManyToManyField(...)

            class Meta:
                verbose_name = _("Master")
    """

    class Meta:
        abstract = True

Appointment mixins

codex_django.booking.mixins.appointment

codex_django.booking.mixins.appointment

Composable mixins for the appointment/booking record model.

Usage::

from codex_django.booking.mixins import AbstractBookableAppointment

class Appointment(AbstractBookableAppointment):
    master = models.ForeignKey("masters.Master", ...)
    service = models.ForeignKey("services.Service", ...)
    client = models.ForeignKey(settings.AUTH_USER_MODEL, ...)

    class Meta:
        verbose_name = _("Appointment")

FK fields are NOT included — the user defines them pointing at their own Master/Service/Client models.

Classes

AppointmentStatusMixin

Bases: Model

Appointment lifecycle status.

Status constants are class-level attributes so that the adapter and selectors can reference them without hardcoding strings::

Appointment.STATUS_PENDING
Appointment.STATUS_CONFIRMED

Admin fieldsets example::

(_("Status"), {
    "fields": ("status",),
}),

Source code in src/codex_django/booking/mixins/appointment.py

class AppointmentStatusMixin(models.Model):
    """Appointment lifecycle status.

    Status constants are class-level attributes so that the adapter
    and selectors can reference them without hardcoding strings::

        Appointment.STATUS_PENDING
        Appointment.STATUS_CONFIRMED

    Admin fieldsets example::

        (_("Status"), {
            "fields": ("status",),
        }),
    """

    STATUS_PENDING = "pending"
    STATUS_CONFIRMED = "confirmed"
    STATUS_CANCELLED = "cancelled"
    STATUS_COMPLETED = "completed"
    STATUS_NO_SHOW = "no_show"
    STATUS_RESCHEDULE_PROPOSED = "reschedule_proposed"

    STATUS_CHOICES = [
        (STATUS_PENDING, _("Pending")),
        (STATUS_CONFIRMED, _("Confirmed")),
        (STATUS_CANCELLED, _("Cancelled")),
        (STATUS_COMPLETED, _("Completed")),
        (STATUS_NO_SHOW, _("No Show")),
        (STATUS_RESCHEDULE_PROPOSED, _("Reschedule Proposed")),
    ]

    ACTIVE_STATUSES = [STATUS_PENDING, STATUS_CONFIRMED, STATUS_RESCHEDULE_PROPOSED]

    status = models.CharField(
        _("Status"),
        max_length=30,
        choices=STATUS_CHOICES,
        default=STATUS_PENDING,
        db_index=True,
    )

    class Meta:
        abstract = True

AppointmentCoreMixin

Bases: Model

Core booking data: when and how long.

Admin fieldsets example::

(_("Booking Details"), {
    "fields": ("datetime_start", "duration_minutes"),
}),

Source code in src/codex_django/booking/mixins/appointment.py

class AppointmentCoreMixin(models.Model):
    """Core booking data: when and how long.

    Admin fieldsets example::

        (_("Booking Details"), {
            "fields": ("datetime_start", "duration_minutes"),
        }),
    """

    datetime_start = models.DateTimeField(
        _("Start Time"),
        db_index=True,
    )
    duration_minutes = models.PositiveIntegerField(
        _("Duration (min)"),
    )

    class Meta:
        abstract = True

AbstractBookableAppointment

Bases: AppointmentStatusMixin, AppointmentCoreMixin, Model

Convenience base that assembles all appointment mixins.

Usage::

class Appointment(AbstractBookableAppointment):
    master = models.ForeignKey("masters.Master", on_delete=models.CASCADE)
    service = models.ForeignKey("services.Service", on_delete=models.CASCADE)
    client = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)

    class Meta:
        verbose_name = _("Appointment")
        ordering = ["-datetime_start"]

Source code in src/codex_django/booking/mixins/appointment.py

class AbstractBookableAppointment(
    AppointmentStatusMixin,
    AppointmentCoreMixin,
    models.Model,
):
    """Convenience base that assembles all appointment mixins.

    Usage::

        class Appointment(AbstractBookableAppointment):
            master = models.ForeignKey("masters.Master", on_delete=models.CASCADE)
            service = models.ForeignKey("services.Service", on_delete=models.CASCADE)
            client = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)

            class Meta:
                verbose_name = _("Appointment")
                ordering = ["-datetime_start"]
    """

    class Meta:
        abstract = True

Schedule mixins

codex_django.booking.mixins.schedule

codex_django.booking.mixins.schedule

Relational models for master working schedule and days off.

AbstractWorkingDay replaces JSONField work_days — gives proper SQL filtering, per-day hours, and index support.

Usage::

from codex_django.booking.mixins import AbstractWorkingDay, MasterDayOffMixin

class MasterWorkingDay(AbstractWorkingDay):
    master = models.ForeignKey(
        "masters.Master",
        on_delete=models.CASCADE,
        related_name="working_days",
    )

    class Meta:
        verbose_name = _("Working Day")
        unique_together = [("master", "weekday")]

class MasterDayOff(MasterDayOffMixin):
    master = models.ForeignKey(
        "masters.Master",
        on_delete=models.CASCADE,
        related_name="days_off",
    )

    class Meta:
        verbose_name = _("Day Off")
        unique_together = [("master", "date")]

Classes

AbstractWorkingDay

Bases: Model

Per-weekday schedule for a master.

FK to the master model is NOT included — the user adds it pointing at their own model.

Admin fieldsets example::

(_("Schedule"), {
    "fields": ("weekday", "start_time", "end_time", "break_start", "break_end"),
}),

Source code in src/codex_django/booking/mixins/schedule.py

class AbstractWorkingDay(models.Model):
    """Per-weekday schedule for a master.

    FK to the master model is NOT included — the user adds it
    pointing at their own model.

    Admin fieldsets example::

        (_("Schedule"), {
            "fields": ("weekday", "start_time", "end_time", "break_start", "break_end"),
        }),
    """

    WEEKDAY_CHOICES = [
        (0, _("Monday")),
        (1, _("Tuesday")),
        (2, _("Wednesday")),
        (3, _("Thursday")),
        (4, _("Friday")),
        (5, _("Saturday")),
        (6, _("Sunday")),
    ]

    weekday = models.PositiveSmallIntegerField(
        _("Day of Week"),
        choices=WEEKDAY_CHOICES,
        validators=[MinValueValidator(0), MaxValueValidator(6)],
        db_index=True,
    )
    start_time = models.TimeField(_("Start Time"))
    end_time = models.TimeField(_("End Time"))
    break_start = models.TimeField(_("Break Start"), null=True, blank=True)
    break_end = models.TimeField(_("Break End"), null=True, blank=True)

    class Meta:
        abstract = True

    def __str__(self) -> str:
        day_name = dict(self.WEEKDAY_CHOICES).get(self.weekday, self.weekday)
        return f"{day_name}: {self.start_time}–{self.end_time}"

MasterDayOffMixin

Bases: Model

A single day-off record for a master.

FK to the master model is NOT included — the user adds it.

Admin fieldsets example::

(_("Day Off"), {
    "fields": ("date", "reason"),
}),

Source code in src/codex_django/booking/mixins/schedule.py

class MasterDayOffMixin(models.Model):
    """A single day-off record for a master.

    FK to the master model is NOT included — the user adds it.

    Admin fieldsets example::

        (_("Day Off"), {
            "fields": ("date", "reason"),
        }),
    """

    date = models.DateField(_("Date"), db_index=True)
    reason = models.CharField(
        _("Reason"),
        max_length=255,
        blank=True,
    )

    class Meta:
        abstract = True

    def __str__(self) -> str:
        return f"Day off: {self.date}"

Settings mixins

codex_django.booking.mixins.settings

codex_django.booking.mixins.settings

Booking system configuration model mixins.

Usage::

from codex_django.booking.mixins import AbstractBookingSettings

class BookingSettings(AbstractBookingSettings):
    class Meta:
        verbose_name = _("Booking Settings")

# In your Django settings:
# CODEX_BOOKING_SETTINGS_MODEL = 'system.BookingSettings'

Classes

BookingSettingsMixin

Bases: Model

Core booking configuration fields.

Admin fieldsets example::

(_("Booking Defaults"), {
    "fields": (
        "step_minutes",
        "default_buffer_between_minutes",
        "min_advance_minutes",
        "max_advance_days",
    ),
}),
(_("Default Working Hours — Weekdays"), {
    "fields": ("work_start_weekdays", "work_end_weekdays"),
}),
(_("Default Working Hours — Saturday"), {
    "fields": ("work_start_saturday", "work_end_saturday"),
    "classes": ("collapse",),
}),

Source code in src/codex_django/booking/mixins/settings.py

class BookingSettingsMixin(models.Model):
    """Core booking configuration fields.

    Admin fieldsets example::

        (_("Booking Defaults"), {
            "fields": (
                "step_minutes",
                "default_buffer_between_minutes",
                "min_advance_minutes",
                "max_advance_days",
            ),
        }),
        (_("Default Working Hours — Weekdays"), {
            "fields": ("work_start_weekdays", "work_end_weekdays"),
        }),
        (_("Default Working Hours — Saturday"), {
            "fields": ("work_start_saturday", "work_end_saturday"),
            "classes": ("collapse",),
        }),
    """

    step_minutes = models.PositiveIntegerField(
        _("Slot Step (min)"),
        default=30,
        help_text=_("Time grid granularity for the booking engine."),
    )
    default_buffer_between_minutes = models.PositiveIntegerField(
        _("Default Buffer Between Appointments (min)"),
        default=0,
    )
    min_advance_minutes = models.PositiveIntegerField(
        _("Minimum Advance Booking (min)"),
        default=60,
        help_text=_("Global minimum; masters can override individually."),
    )
    max_advance_days = models.PositiveIntegerField(
        _("Maximum Advance Booking (days)"),
        default=60,
    )

    # Default working hours (fallback when master has no individual schedule)
    work_start_weekdays = models.TimeField(_("Weekday Start"), null=True, blank=True)
    work_end_weekdays = models.TimeField(_("Weekday End"), null=True, blank=True)
    work_start_saturday = models.TimeField(_("Saturday Start"), null=True, blank=True)
    work_end_saturday = models.TimeField(_("Saturday End"), null=True, blank=True)

    class Meta:
        abstract = True

    def to_dict(self) -> dict[str, Any]:
        """Serialize concrete fields for Redis storage."""
        data: dict[str, Any] = {}
        for field in self._meta.get_fields():
            if field.concrete and not field.many_to_many and not field.one_to_many:
                if field.name in ("id", "pk"):
                    continue
                value = getattr(self, field.name)
                data[field.name] = str(value) if value is not None else None
        return data

Functions

to_dict()

Serialize concrete fields for Redis storage.

Source code in src/codex_django/booking/mixins/settings.py

def to_dict(self) -> dict[str, Any]:
    """Serialize concrete fields for Redis storage."""
    data: dict[str, Any] = {}
    for field in self._meta.get_fields():
        if field.concrete and not field.many_to_many and not field.one_to_many:
            if field.name in ("id", "pk"):
                continue
            value = getattr(self, field.name)
            data[field.name] = str(value) if value is not None else None
    return data

BookingSettingsSyncMixin

Bases: LifecycleModelMixin, Model

Sync booking settings to Redis on save.

Follows rule R1: Redis errors are swallowed with logging so that a Redis outage never prevents saving settings in the database.

Source code in src/codex_django/booking/mixins/settings.py

class BookingSettingsSyncMixin(LifecycleModelMixin, models.Model):
    """Sync booking settings to Redis on save.

    Follows rule R1: Redis errors are swallowed with logging so that
    a Redis outage never prevents saving settings in the database.
    """

    class Meta:
        abstract = True

    @hook(AFTER_SAVE)  # type: ignore[untyped-decorator]
    def sync_booking_settings_to_redis(self) -> None:
        """Persist the latest booking settings payload to Redis after save."""
        from django.conf import settings as django_settings

        if django_settings.DEBUG and not getattr(django_settings, "CODEX_REDIS_ENABLED", False):
            return

        try:
            from codex_django.core.redis.managers.booking import (
                get_booking_cache_manager,
            )

            manager = get_booking_cache_manager()
            if hasattr(self, "to_dict"):
                data = self.to_dict()
                if data:
                    key = manager.make_key("settings")
                    from asgiref.sync import async_to_sync

                    async_to_sync(manager.string.set)(key, str(data))
        except Exception:
            log.warning("Failed to sync booking settings to Redis", exc_info=True)

Functions

sync_booking_settings_to_redis()

Persist the latest booking settings payload to Redis after save.

Source code in src/codex_django/booking/mixins/settings.py

@hook(AFTER_SAVE)  # type: ignore[untyped-decorator]
def sync_booking_settings_to_redis(self) -> None:
    """Persist the latest booking settings payload to Redis after save."""
    from django.conf import settings as django_settings

    if django_settings.DEBUG and not getattr(django_settings, "CODEX_REDIS_ENABLED", False):
        return

    try:
        from codex_django.core.redis.managers.booking import (
            get_booking_cache_manager,
        )

        manager = get_booking_cache_manager()
        if hasattr(self, "to_dict"):
            data = self.to_dict()
            if data:
                key = manager.make_key("settings")
                from asgiref.sync import async_to_sync

                async_to_sync(manager.string.set)(key, str(data))
    except Exception:
        log.warning("Failed to sync booking settings to Redis", exc_info=True)

AbstractBookingSettings

Bases: BookingSettingsMixin, BookingSettingsSyncMixin, Model

Convenience base that assembles settings fields + Redis sync.

Usage::

class BookingSettings(AbstractBookingSettings):
    class Meta:
        verbose_name = _("Booking Settings")

Source code in src/codex_django/booking/mixins/settings.py

class AbstractBookingSettings(
    BookingSettingsMixin,
    BookingSettingsSyncMixin,
    models.Model,
):
    """Convenience base that assembles settings fields + Redis sync.

    Usage::

        class BookingSettings(AbstractBookingSettings):
            class Meta:
                verbose_name = _("Booking Settings")
    """

    class Meta:
        abstract = True