Code reference › domain › schedule

module

pretalx.schedule.domain.changes

Functions 5

calculate_schedule_changes(schedule) -> dictsource

def calculate_schedule_changes(schedule) -> dict:
    result = {
        "count": 0,
        "action": "update",
        "new_talks": [],
        "canceled_talks": [],
        "moved_talks": [],
    }

    if not schedule.previous_schedule:
        result["action"] = "create"
        return result

    class Slot(NamedTuple):
        submission: object
        room: object
        local_start: object

    old_slots = {
        Slot(slot.submission, slot.room, slot.local_start): slot
        for slot in schedule.previous_schedule.scheduled_talks
    }
    new_slots = {
        Slot(slot.submission, slot.room, slot.local_start): slot
        for slot in schedule.scheduled_talks
    }

    old_slot_set = set(old_slots.keys())
    new_slot_set = set(new_slots.keys())
    old_submissions = {slot.submission for slot in old_slots}
    new_submissions = {slot.submission for slot in new_slots}
    handled_submissions = set()
    new_by_submission = defaultdict(list)
    old_by_submission = defaultdict(list)

    for slot in new_slot_set:
        new_by_submission[slot.submission].append(new_slots[slot])
    for slot in old_slot_set:
        old_by_submission[slot.submission].append(old_slots[slot])

    moved_or_missing = old_slot_set - new_slot_set - {None}
    moved_or_new = new_slot_set - old_slot_set - {None}

    for entry in moved_or_missing:
        if entry.submission in handled_submissions:
            continue
        if entry.submission not in new_submissions:
            result["canceled_talks"] += old_by_submission[entry.submission]
        else:
            new, canceled, moved = _handle_submission_move(
                entry.submission, old_slots, new_slots
            )
            result["new_talks"] += new
            result["canceled_talks"] += canceled
            result["moved_talks"] += moved
        handled_submissions.add(entry.submission)

    for entry in moved_or_new:
        if entry.submission in handled_submissions:
            continue
        if entry.submission not in old_submissions:
            result["new_talks"] += new_by_submission[entry.submission]
        else:
            new, canceled, moved = _handle_submission_move(
                entry.submission, old_slots, new_slots
            )
            result["new_talks"] += new
            result["canceled_talks"] += canceled
            result["moved_talks"] += moved
        handled_submissions.add(entry.submission)

    result["count"] = (
        len(result["new_talks"])
        + len(result["canceled_talks"])
        + len(result["moved_talks"])
    )

    return result

get_cached_schedule_changes(schedule) -> dictsource

def get_cached_schedule_changes(schedule) -> dict:
    cache_key = f"schedule_{schedule.id}_changes"
    cached_data = schedule.event.cache.get(cache_key)

    if cached_data:
        with suppress(json.JSONDecodeError, KeyError):
            serialized = json.loads(cached_data)
            return _deserialize_changes(serialized, schedule.event)

    result = calculate_schedule_changes(schedule)

    # WIP schedules: 60 seconds cache; released schedules: 10 minutes.
    timeout = 60 if schedule.version is None else 600

    with suppress(Exception):
        serialized = _serialize_changes(result)
        schedule.event.cache.set(cache_key, json.dumps(serialized), timeout)

    if schedule.version is None:
        update_unreleased_schedule_changes(
            schedule.event, _get_boolean_changes(schedule, result)
        )

    return result

has_unreleased_schedule_changes(event) -> boolsourceCompute whether event has unreleased WIP schedule changes.

Prefer event.has_unreleased_schedule_changes (cached_property on Event); it works better in tests and on instances with a broken or slow cache, as it additionally caches on the event instance.

def has_unreleased_schedule_changes(event) -> bool:
    """Compute whether ``event`` has unreleased WIP schedule changes.

    Prefer ``event.has_unreleased_schedule_changes`` (cached_property on
    Event); it works better in tests and on instances with a broken or
    slow cache, as it additionally caches on the event instance."""
    cache_key = "has_unreleased_schedule_changes"
    cached_value = event.cache.get(cache_key)

    if cached_value is not None:
        return cached_value

    with scope(event=event):
        value = _get_boolean_changes(event.wip_schedule)
    update_unreleased_schedule_changes(event, value)
    return value

invalidate_cached_schedule_changes(schedule)source

def invalidate_cached_schedule_changes(schedule):
    cache_key = f"schedule_{schedule.id}_changes"
    schedule.event.cache.delete(cache_key)

update_unreleased_schedule_changes(event, value=None)source

def update_unreleased_schedule_changes(event, value=None):
    cache_key = "has_unreleased_schedule_changes"
    if value is None:
        invalidate_cached_schedule_changes(event.wip_schedule)
        value = _get_boolean_changes(event.wip_schedule)
    event.cache.set(cache_key, value, 24 * 60 * 60)