Code reference › domain › schedule

module

pretalx.schedule.domain.availability

Functions 2

merged_speaker_availabilities(schedule)sourceReturns a {slot_pk: [Availability, ...]} dict.

One pass for the full schedule rather than per-session access to cut down on queries.

def merged_speaker_availabilities(schedule):
    """Returns a ``{slot_pk: [Availability, ...]}`` dict.

    One pass for the full schedule rather than per-session access
    to cut down on queries.
    """
    event = schedule.event
    speaker_avails = collections.defaultdict(list)
    for avail in event.valid_availabilities.filter(person__isnull=False):
        speaker_avails[avail.person_id].append(avail)

    result = {}
    talks = (
        schedule.talks.filter(submission__isnull=False)
        .select_related("submission")
        .with_sorted_speakers()
    )
    for talk in talks:
        speaker_sets = [
            speaker_avails[speaker.pk]
            for speaker in talk.submission.sorted_speakers
            if speaker_avails[speaker.pk]
        ]
        if len(speaker_sets) == 1:
            result[talk.id] = Availability.union(speaker_sets[0])
        elif speaker_sets:
            result[talk.id] = Availability.intersection(*speaker_sets)
        else:
            result[talk.id] = []
    return result

replace_availabilities(instance, availabilities)sourceReplace all availabilities for instance (a Room or SpeakerProfile) with the given list, in a single transaction.

def replace_availabilities(instance, availabilities):
    """Replace all availabilities for ``instance`` (a Room or SpeakerProfile)
    with the given list, in a single transaction."""
    merged = Availability.union(availabilities)
    field_name = instance.availabilities.field.name
    for avail in merged:
        setattr(avail, field_name, instance)

    with transaction.atomic():
        instance.availabilities.all().delete()
        Availability.objects.bulk_create(merged)