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_resources() 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_resource_id=None, resource_selections=None, mode=BookingMode.SINGLE_DAY, overlap_allowed=False, parallel_groups=None, max_solutions=None, 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_resource_id	int | None	Optional resource id that constrains the search.	None
resource_selections	ResourceSelections	Optional per-service resource 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 | None	Maximum number of engine solutions to compute. Defaults to 50 for single-service requests and 2000 for multi-service chains.	None
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_resource_id: int | None = None,
    resource_selections: ResourceSelections = None,
    mode: BookingMode = BookingMode.SINGLE_DAY,
    overlap_allowed: bool = False,
    parallel_groups: dict[int, str] | None = None,
    max_solutions: int | None = None,
    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_resource_id: Optional resource id that constrains the search.
        resource_selections: Optional per-service resource 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. Defaults
            to 50 for single-service requests and 2000 for multi-service chains.
        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_resource_id=locked_resource_id,
        resource_selections=resource_selections,
        mode=mode,
        overlap_allowed=overlap_allowed,
        parallel_groups=parallel_groups,
    )

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

    availability = adapter.build_resources_availability(
        resource_ids=unique_resource_ids,
        target_date=target_date,
        cache_ttl=cache_ttl,
    )

    effective_max_solutions = _get_effective_max_solutions(service_ids, max_solutions)

    finder = ChainFinder(step_minutes=adapter.step_minutes)
    return finder.find(
        request=request,
        resources_availability=availability,
        max_solutions=effective_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,
    )

build_month_grid_cells(year, month, visible_days)

Expand visible month days into a 7-column calendar grid.

The cabinet date-time picker expects a Monday-first month grid. Some upstream calendar payloads expose only the visible days for the current month, which makes the first week shift left when the month starts on a later weekday. This helper pads the visible days with leading and trailing blank cells so templates can render a stable 7-column matrix.

Parameters:

Name	Type	Description	Default
year	int	Target calendar year.	required
month	int	Target calendar month.	required
visible_days	list[dict[str, Any]]	Flat list of day payloads for the given month.	required

Returns:

Type	Description
list[dict[str, Any]]	A list of calendar cell dicts. Blank placeholders include
list[dict[str, Any]]	{"blank": True}; real day payloads are copied and annotated with
list[dict[str, Any]]	blank=False.

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

def build_month_grid_cells(year: int, month: int, visible_days: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Expand visible month days into a 7-column calendar grid.

    The cabinet date-time picker expects a Monday-first month grid. Some
    upstream calendar payloads expose only the visible days for the current
    month, which makes the first week shift left when the month starts on a
    later weekday. This helper pads the visible days with leading and trailing
    blank cells so templates can render a stable 7-column matrix.

    Args:
        year: Target calendar year.
        month: Target calendar month.
        visible_days: Flat list of day payloads for the given month.

    Returns:
        A list of calendar cell dicts. Blank placeholders include
        ``{"blank": True}``; real day payloads are copied and annotated with
        ``blank=False``.
    """
    if not visible_days:
        return []

    first_weekday, month_days = monthrange(year, month)
    month_payload = visible_days[:month_days]
    leading_blanks = first_weekday
    trailing_blanks = (-((leading_blanks + len(month_payload)) % 7)) % 7

    cells: list[dict[str, Any]] = [{"blank": True} for _ in range(leading_blanks)]
    cells.extend({**day, "blank": False} for day in month_payload)
    cells.extend({"blank": True} for _ in range(trailing_blanks))
    return cells

build_picker_day_rows(*, start_date, horizon, available_dates, has_service_scope=True)

Build booking day-picker rows with month headers and blank paddings.

This helper mirrors the cabinet date-picker payload shape used by booking workflows. It keeps month grouping and leading blank cell generation in one reusable place so project services don't reimplement this rendering contract.

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

def build_picker_day_rows(
    *,
    start_date: date,
    horizon: int,
    available_dates: set[str],
    has_service_scope: bool = True,
) -> list[dict[str, str | int | bool]]:
    """Build booking day-picker rows with month headers and blank paddings.

    This helper mirrors the cabinet date-picker payload shape used by booking
    workflows. It keeps month grouping and leading blank cell generation in one
    reusable place so project services don't reimplement this rendering contract.
    """
    rows: list[dict[str, str | int | bool]] = []
    if horizon <= 0:
        return rows

    current_month_key: str | None = None
    for offset in range(horizon):
        current_date = start_date.fromordinal(start_date.toordinal() + offset)
        month_key = current_date.strftime("%Y-%m")
        month_label = current_date.strftime("%B %Y")
        if month_key != current_month_key:
            for blank_index in range(current_date.weekday()):
                rows.append(
                    {
                        "day": "",
                        "iso": f"{month_key}-blank-{blank_index}",
                        "busy": True,
                        "available": False,
                        "label": "",
                        "month_key": month_key,
                        "month_label": month_label,
                    }
                )
            current_month_key = month_key

        iso = current_date.isoformat()
        is_available = iso in available_dates if has_service_scope else False
        rows.append(
            {
                "day": current_date.day,
                "iso": iso,
                "busy": has_service_scope and not is_available,
                "available": is_available,
                "label": current_date.strftime("%b %d, %Y"),
                "month_key": month_key,
                "month_label": month_label,
            }
        )
    return rows

parse_resource_selections(raw_value)

Parse serialized resource selections from HTTP payloads.

Accepts a JSON object and drops non-selections (any, empty, null). Returns None when the payload is missing, invalid, or produces no effective selections.

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

def parse_resource_selections(raw_value: str | None) -> dict[str, str] | None:
    """Parse serialized resource selections from HTTP payloads.

    Accepts a JSON object and drops non-selections (``any``, empty, ``null``).
    Returns ``None`` when the payload is missing, invalid, or produces no
    effective selections.
    """
    if not raw_value:
        return None
    try:
        parsed = json.loads(raw_value)
    except Exception:
        return None
    if not isinstance(parsed, dict):
        return None

    normalized: dict[str, str] = {}
    for key, value in parsed.items():
        if value in ("", None, "any"):
            continue
        normalized[str(key)] = str(value)
    return normalized or None

normalize_slot_payload(payload)

Normalize slot payloads from booking gateway calls.

Supports engine results (get_unique_start_times), dict payloads like {slot: allowed}, and plain iterables of slot strings.

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

def normalize_slot_payload(payload: Any) -> list[str]:
    """Normalize slot payloads from booking gateway calls.

    Supports engine results (``get_unique_start_times``), ``dict`` payloads
    like ``{slot: allowed}``, and plain iterables of slot strings.
    """
    if payload is None:
        return []

    get_unique_start_times = getattr(payload, "get_unique_start_times", None)
    if callable(get_unique_start_times):
        try:
            return sorted({str(slot) for slot in get_unique_start_times() if str(slot)})
        except Exception:
            return []

    if isinstance(payload, Mapping):
        return sorted(str(slot) for slot, allowed in payload.items() if allowed)

    if isinstance(payload, list | tuple | set):
        return sorted(str(slot) for slot in payload if str(slot))

    return []

get_nearest_slots(adapter, service_ids, search_from, *, locked_resource_id=None, resource_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_resource_id	int | None	Optional resource id that constrains the search.	None
resource_selections	ResourceSelections	Optional per-service resource 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_resource_id: int | None = None,
    resource_selections: ResourceSelections = 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_resource_id: Optional resource id that constrains the search.
        resource_selections: Optional per-service resource 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_resource_id=locked_resource_id,
        resource_selections=resource_selections,
        mode=mode,
        overlap_allowed=overlap_allowed,
        parallel_groups=parallel_groups,
    )

    def get_availability_for_date(
        d: date,
    ) -> dict[str, Any]:
        all_resource_ids: list[int] = []
        for sr in request.service_requests:
            all_resource_ids.extend(int(mid) for mid in sr.possible_resource_ids)
        return adapter.build_resources_availability(
            resource_ids=list(set(all_resource_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, resource_id, client, extra_fields=None, resource_selections=None, mode=BookingMode.SINGLE_DAY, overlap_allowed=False, parallel_groups=None, max_solutions=None, persistence_hook=None)

Create a booking with concurrency protection.

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

Chain mode: 1. Requires a persistence_hook 2. Locks all candidate resources 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
resource_id	int | None	Resource 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
resource_selections	ResourceSelections	Optional per-service resource 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 | None	Optional cap for engine solutions during final revalidation. Defaults match get_available_slots().	None
persistence_hook	BookingPersistenceHook | None	Required persistence hook for multi-service mode.	None

Returns:

Type	Description
Any | list[Any]	A single appointment instance in locked-resource mode, or a list of
Any | list[Any]	created appointment-like objects in chain 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,
    resource_id: int | None,
    client: Any,
    extra_fields: dict[str, Any] | None = None,
    resource_selections: ResourceSelections = None,
    mode: BookingMode = BookingMode.SINGLE_DAY,
    overlap_allowed: bool = False,
    parallel_groups: dict[int, str] | None = None,
    max_solutions: int | None = None,
    persistence_hook: BookingPersistenceHook | None = None,
) -> Any | list[Any]:
    """Create a booking with concurrency protection.

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

    Chain mode:
    1. Requires a ``persistence_hook``
    2. Locks all candidate resources 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.
        resource_id: Resource chosen for single-service mode.
        client: Client object attached to the booking.
        extra_fields: Optional extra model fields passed to persistence.
        resource_selections: Optional per-service resource 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: Optional cap for engine solutions during final
            revalidation. Defaults match ``get_available_slots()``.
        persistence_hook: Required persistence hook for multi-service mode.

    Returns:
        A single appointment instance in locked-resource mode, or a list of
        created appointment-like objects in chain 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_chain_flow = len(service_ids) > 1 or resource_id is None

    with transaction.atomic():
        if is_chain_flow:
            if len(service_ids) > 1 and 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_resource_id=resource_id,
                resource_selections=resource_selections,
                mode=mode,
                overlap_allowed=overlap_allowed,
                parallel_groups=parallel_groups,
            )
            all_resource_ids: set[int] = {
                int(mid) for sr in request.service_requests for mid in sr.possible_resource_ids
            }
            adapter.lock_resources(sorted(all_resource_ids))

            result = get_available_slots(
                adapter=adapter,
                service_ids=service_ids,
                target_date=target_date,
                locked_resource_id=resource_id,
                resource_selections=resource_selections,
                mode=mode,
                overlap_allowed=overlap_allowed,
                parallel_groups=parallel_groups,
                max_solutions=max_solutions,
                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.")

            selected_solution = _select_solution_for_time(result, selected_time)
            if selected_solution is None:
                raise SlotAlreadyBookedError("No solution found.")

            if persistence_hook is not None:
                created_appointments = persistence_hook.persist_chain(
                    solution=selected_solution,
                    service_ids=service_ids,
                    client=client,
                    extra_fields=extra_fields,
                )
            else:
                # Generic fallback for single-service "any resource" flows.
                primary_item = next(iter(getattr(selected_solution, "items", [])), None)
                if primary_item is None:
                    raise SlotAlreadyBookedError("No solution found.")
                created_appointments = [
                    _create_single_appointment(
                        appointment_model=appointment_model,
                        resource_id=int(primary_item.resource_id),
                        service_ids=service_ids,
                        starts_at=selected_solution.starts_at,
                        span_minutes=selected_solution.span_minutes,
                        client=client,
                        extra_fields=extra_fields,
                    )
                ]

            chain_resource_ids = {
                str(item.resource_id)
                for item in getattr(selected_solution, "items", [])
                if getattr(item, "resource_id", None)
            }
            if not chain_resource_ids and resource_id is not None:
                chain_resource_ids = {str(resource_id)}

            _date_str = str(target_date)

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

            transaction.on_commit(_invalidate_many)
            return created_appointments

        if resource_id is None:
            raise SlotAlreadyBookedError("Resource id is required for single-resource booking flow.")

        adapter.lock_resources([resource_id])

        result = get_available_slots(
            adapter=adapter,
            service_ids=service_ids,
            target_date=target_date,
            locked_resource_id=resource_id,
            resource_selections=resource_selections,
            mode=mode,
            overlap_allowed=overlap_allowed,
            parallel_groups=parallel_groups,
            max_solutions=max_solutions,
            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.
        selected_solution = _select_solution_for_time(result, selected_time)
        if selected_solution is None:
            raise SlotAlreadyBookedError("No solution found.")

        appointment = _create_single_appointment(
            appointment_model=appointment_model,
            resource_id=resource_id,
            service_ids=service_ids,
            starts_at=selected_solution.starts_at,
            span_minutes=selected_solution.span_minutes,
            client=client,
            extra_fields=extra_fields,
        )

        # R1: invalidate cache AFTER transaction commits
        _resource_id_str = str(resource_id)
        _date_str = str(target_date)
        transaction.on_commit(lambda: cache_adapter.invalidate_master_date(_resource_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_resources(). 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
resource_model	type[Any]	Django model class for resources/executors.	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
step_minutes	int	Time grid step for the engine.	30
timezone	str	Default timezone for booking calculations when no resource timezone is provided.	'UTC'
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:
        resource_model: Django model class for resources/executors.
        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.
        step_minutes: Time grid step for the engine.
        timezone: Default timezone for booking calculations when no resource
            timezone is provided.
        appointment_status_filter: Which statuses count as "busy".
        cache_adapter: Optional cache adapter; defaults to BookingCacheAdapter.
    """

    def __init__(
        self,
        resource_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],
        step_minutes: int = 30,
        timezone: str = "UTC",
        appointment_status_filter: list[str] | None = None,
        cache_adapter: BookingCacheAdapter | None = None,
    ) -> None:
        self.resource_model = resource_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.step_minutes = step_minutes
        self.timezone = timezone
        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

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

    def build_engine_request(
        self,
        service_ids: list[int],
        target_date: date,
        locked_resource_id: int | None = None,
        resource_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/resource data."""
        services = self.service_model.objects.filter(id__in=service_ids).select_related("category")
        service_map = {s.id: s for s in services}
        normalized_resource_selections = self._normalize_resource_selections(service_ids, resource_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_resource_ids(
                service=service,
                weekday=weekday,
                locked_resource_id=locked_resource_id,
                resource_selections=normalized_resource_selections,
                service_id=svc_id,
            )
            possible_ids = self.prioritize_resource_ids(
                resource_ids=possible_ids,
                service=service,
                target_date=target_date,
                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_resources_availability(
        self,
        resource_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 resources entirely
        day_off_ids = set(
            self.day_off_model.objects.filter(master_id__in=resource_ids, date=target_date).values_list(
                "master_id", flat=True
            )
        )

        # Busy intervals (from appointments)
        busy_by_resource = self._get_busy_intervals(resource_ids, target_date, exclude_appointment_ids)

        resources = self.resource_model.objects.filter(pk__in=resource_ids)
        for master in resources:
            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_resource.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

    def prioritize_resource_ids(
        self,
        *,
        resource_ids: list[str],
        service: Any,
        target_date: date,
        service_id: int,
    ) -> list[str]:
        """Public seam for project-specific resource ordering policies."""
        del service, target_date, service_id
        return resource_ids

    # ------------------------------------------------------------------
    # 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 booking settings 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. Booking-level defaults
            settings = self._get_booking_settings()
            day_schedule = self._get_default_day_schedule(settings, weekday)
            if day_schedule is not None:
                start_t, end_t = day_schedule

        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_resources(self, resource_ids: list[int]) -> None:
        """Acquire row-level locks on resource records.

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

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

    def _get_busy_intervals(
        self,
        resource_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 resource_ids}

        appt_filter = {
            "master_id__in": resource_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_resource_ids(
        self,
        service: Any,
        weekday: int,
        locked_resource_id: int | None,
        resource_selections: dict[int, int | None] | None,
        service_id: int,
    ) -> list[str]:
        """Determine which resource IDs can perform a service."""
        if locked_resource_id:
            return [str(locked_resource_id)]

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

        # All active resources for this service's category working on this weekday
        masters = self.resource_model.objects.filter(
            categories=service.category,
            status=self.resource_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_resource_selections(
        self,
        service_ids: list[int],
        resource_selections: dict[str, str] | dict[int, int | None] | None,
    ) -> dict[int, int | None]:
        """Normalize legacy and new resource selection formats.

        Supported input formats:
        - Legacy positional: {"0": "10", "1": "12"}
        - Service keyed: {5: 10, 7: None} or {"5": "10", "7": "12"}
        """
        if not resource_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 resource_selections
        )
        if is_legacy_positional:
            normalized_legacy: dict[int, int | None] = {}
            for key, raw_val in resource_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 resource_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 resource timezone, falling back to the adapter timezone."""
        tz_name = getattr(master, "timezone", None)
        if tz_name == "":
            tz_name = None
        if not tz_name:
            tz_name = self.timezone or "UTC"
        try:
            return zoneinfo.ZoneInfo(tz_name)
        except Exception:
            return zoneinfo.ZoneInfo("UTC")

    def _get_default_day_schedule(self, settings: Any, weekday: int) -> tuple[Any, Any] | None:
        """Return booking-level fallback hours for a weekday."""
        get_day_schedule = getattr(settings, "get_day_schedule", None)
        if callable(get_day_schedule):
            result = get_day_schedule(weekday)
            if isinstance(result, tuple) and len(result) == 2:
                return result
            if result is None:
                return None

        day_names = (
            "monday",
            "tuesday",
            "wednesday",
            "thursday",
            "friday",
            "saturday",
            "sunday",
        )
        day_name = day_names[weekday]
        if getattr(settings, f"{day_name}_is_closed", False):
            return None

        start_t = getattr(settings, f"work_start_{day_name}", None)
        end_t = getattr(settings, f"work_end_{day_name}", None)
        if not start_t or not end_t:
            return None
        return (start_t, end_t)

    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

Functions

build_engine_request(service_ids, target_date, locked_resource_id=None, resource_selections=None, mode=BookingMode.SINGLE_DAY, overlap_allowed=False, parallel_groups=None)

Build a BookingEngineRequest from DB service/resource data.

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

def build_engine_request(
    self,
    service_ids: list[int],
    target_date: date,
    locked_resource_id: int | None = None,
    resource_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/resource data."""
    services = self.service_model.objects.filter(id__in=service_ids).select_related("category")
    service_map = {s.id: s for s in services}
    normalized_resource_selections = self._normalize_resource_selections(service_ids, resource_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_resource_ids(
            service=service,
            weekday=weekday,
            locked_resource_id=locked_resource_id,
            resource_selections=normalized_resource_selections,
            service_id=svc_id,
        )
        possible_ids = self.prioritize_resource_ids(
            resource_ids=possible_ids,
            service=service,
            target_date=target_date,
            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_resources_availability(resource_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_resources_availability(
    self,
    resource_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 resources entirely
    day_off_ids = set(
        self.day_off_model.objects.filter(master_id__in=resource_ids, date=target_date).values_list(
            "master_id", flat=True
        )
    )

    # Busy intervals (from appointments)
    busy_by_resource = self._get_busy_intervals(resource_ids, target_date, exclude_appointment_ids)

    resources = self.resource_model.objects.filter(pk__in=resource_ids)
    for master in resources:
        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_resource.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

prioritize_resource_ids(*, resource_ids, service, target_date, service_id)

Public seam for project-specific resource ordering policies.

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

def prioritize_resource_ids(
    self,
    *,
    resource_ids: list[str],
    service: Any,
    target_date: date,
    service_id: int,
) -> list[str]:
    """Public seam for project-specific resource ordering policies."""
    del service, target_date, service_id
    return resource_ids

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 booking settings 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 booking settings 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. Booking-level defaults
        settings = self._get_booking_settings()
        day_schedule = self._get_default_day_schedule(settings, weekday)
        if day_schedule is not None:
            start_t, end_t = day_schedule

    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_resources(resource_ids)

Acquire row-level locks on resource 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_resources(self, resource_ids: list[int]) -> None:
    """Acquire row-level locks on resource records.

    Must be called inside ``transaction.atomic()``.
    IDs are sorted to prevent deadlocks.
    """
    if not resource_ids:
        return
    sorted_ids = sorted(resource_ids)
    list(self.resource_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 the booking adapter 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 the booking adapter 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

Note: The module keeps historical master naming for model-level compatibility. Runtime selector/gateway contracts use resource_* naming.

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",
    ),
}),
(_("Monday"), {"fields": ("monday_is_closed", "work_start_monday", "work_end_monday")}),
(_("Tuesday"), {"fields": ("tuesday_is_closed", "work_start_tuesday", "work_end_tuesday")}),
(_("Wednesday"), {"fields": ("wednesday_is_closed", "work_start_wednesday", "work_end_wednesday")}),
(_("Thursday"), {"fields": ("thursday_is_closed", "work_start_thursday", "work_end_thursday")}),
(_("Friday"), {"fields": ("friday_is_closed", "work_start_friday", "work_end_friday")}),
(_("Saturday"), {"fields": ("saturday_is_closed", "work_start_saturday", "work_end_saturday")}),
(_("Sunday"), {"fields": ("sunday_is_closed", "work_start_sunday", "work_end_sunday")}),

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",
            ),
        }),
        (_("Monday"), {"fields": ("monday_is_closed", "work_start_monday", "work_end_monday")}),
        (_("Tuesday"), {"fields": ("tuesday_is_closed", "work_start_tuesday", "work_end_tuesday")}),
        (_("Wednesday"), {"fields": ("wednesday_is_closed", "work_start_wednesday", "work_end_wednesday")}),
        (_("Thursday"), {"fields": ("thursday_is_closed", "work_start_thursday", "work_end_thursday")}),
        (_("Friday"), {"fields": ("friday_is_closed", "work_start_friday", "work_end_friday")}),
        (_("Saturday"), {"fields": ("saturday_is_closed", "work_start_saturday", "work_end_saturday")}),
        (_("Sunday"), {"fields": ("sunday_is_closed", "work_start_sunday", "work_end_sunday")}),
    """

    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 a resource has no individual schedule)
    monday_is_closed = models.BooleanField(_("Monday Closed"), default=False)
    work_start_monday = models.TimeField(_("Monday Start"), null=True, blank=True)
    work_end_monday = models.TimeField(_("Monday End"), null=True, blank=True)
    tuesday_is_closed = models.BooleanField(_("Tuesday Closed"), default=False)
    work_start_tuesday = models.TimeField(_("Tuesday Start"), null=True, blank=True)
    work_end_tuesday = models.TimeField(_("Tuesday End"), null=True, blank=True)
    wednesday_is_closed = models.BooleanField(_("Wednesday Closed"), default=False)
    work_start_wednesday = models.TimeField(_("Wednesday Start"), null=True, blank=True)
    work_end_wednesday = models.TimeField(_("Wednesday End"), null=True, blank=True)
    thursday_is_closed = models.BooleanField(_("Thursday Closed"), default=False)
    work_start_thursday = models.TimeField(_("Thursday Start"), null=True, blank=True)
    work_end_thursday = models.TimeField(_("Thursday End"), null=True, blank=True)
    friday_is_closed = models.BooleanField(_("Friday Closed"), default=False)
    work_start_friday = models.TimeField(_("Friday Start"), null=True, blank=True)
    work_end_friday = models.TimeField(_("Friday End"), null=True, blank=True)
    saturday_is_closed = models.BooleanField(_("Saturday Closed"), default=False)
    work_start_saturday = models.TimeField(_("Saturday Start"), null=True, blank=True)
    work_end_saturday = models.TimeField(_("Saturday End"), null=True, blank=True)
    sunday_is_closed = models.BooleanField(_("Sunday Closed"), default=False)
    work_start_sunday = models.TimeField(_("Sunday Start"), null=True, blank=True)
    work_end_sunday = models.TimeField(_("Sunday 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

    def get_day_schedule(self, weekday: int) -> tuple[Any, Any] | None:
        """Return the configured fallback schedule for a weekday.

        Weekday numbering follows ``date.weekday()``:
        ``0=Monday`` ... ``6=Sunday``.
        """
        day_names = (
            "monday",
            "tuesday",
            "wednesday",
            "thursday",
            "friday",
            "saturday",
            "sunday",
        )
        day_name = day_names[weekday]
        if getattr(self, f"{day_name}_is_closed", False):
            return None
        start_t = getattr(self, f"work_start_{day_name}", None)
        end_t = getattr(self, f"work_end_{day_name}", None)
        if not start_t or not end_t:
            return None
        return (start_t, end_t)

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

get_day_schedule(weekday)

Return the configured fallback schedule for a weekday.

Weekday numbering follows date.weekday(): 0=Monday ... 6=Sunday.

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

def get_day_schedule(self, weekday: int) -> tuple[Any, Any] | None:
    """Return the configured fallback schedule for a weekday.

    Weekday numbering follows ``date.weekday()``:
    ``0=Monday`` ... ``6=Sunday``.
    """
    day_names = (
        "monday",
        "tuesday",
        "wednesday",
        "thursday",
        "friday",
        "saturday",
        "sunday",
    )
    day_name = day_names[weekday]
    if getattr(self, f"{day_name}_is_closed", False):
        return None
    start_t = getattr(self, f"work_start_{day_name}", None)
    end_t = getattr(self, f"work_end_{day_name}", None)
    if not start_t or not end_t:
        return None
    return (start_t, end_t)

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")
                    with manager.sync_string() as string:
                        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")
                with manager.sync_string() as string:
                    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