Code reference › domain › person

module

pretalx.person.domain.queries.profile

Functions 9

annotate_speaker_submission_counts(qs, *, event, submissions)sourceAnnotate a SpeakerProfile queryset with submission_count and accepted_submission_count for event.

Pass submissions the user may see, usually submissions_for_user. Need to be passed so that the counts are accurate and not e.g. limited by a pre-existing search on the queryset.

def annotate_speaker_submission_counts(qs, *, event, submissions):
    """Annotate a SpeakerProfile queryset with submission_count and
    accepted_submission_count for ``event``.

    Pass submissions the user may see, usually submissions_for_user.
    Need to be passed so that the counts are accurate and not e.g. limited
    by a pre-existing search on the queryset."""
    event_filter = Q(submissions__event=event) & Q(submissions__in=submissions)
    return qs.annotate(
        submission_count=Count("submissions", filter=event_filter, distinct=True),
        accepted_submission_count=Count(
            "submissions",
            filter=event_filter
            & Q(submissions__state__in=SubmissionStates.accepted_states),
            distinct=True,
        ),
    )

annotate_user_submission_counts(qs, *, events)sourceAnnotate a User queryset with submission_count and accepted_submission_count, scoped to events via the user's profiles.

def annotate_user_submission_counts(qs, *, events):
    """Annotate a User queryset with submission_count and
    accepted_submission_count, scoped to ``events`` via the user's profiles."""
    events_filter = Q(profiles__submissions__event__in=events)
    return qs.annotate(
        submission_count=Count(
            "profiles__submissions", filter=events_filter, distinct=True
        ),
        accepted_submission_count=Count(
            "profiles__submissions",
            filter=events_filter
            & Q(profiles__submissions__state__in=SubmissionStates.accepted_states),
            distinct=True,
        ),
    )

filter_by_accepted_role(qs, role)sourceFilter speakers/users on the accepted_submission_count annotation.

The caller must have annotated accepted_submission_count first (see annotate_speaker_submission_counts / annotate_user_submission_counts).

def filter_by_accepted_role(qs, role):
    """Filter speakers/users on the ``accepted_submission_count`` annotation.
    The caller must have annotated ``accepted_submission_count`` first (see
    ``annotate_speaker_submission_counts`` / ``annotate_user_submission_counts``).
    """
    if role == "speaker":
        return qs.filter(accepted_submission_count__gt=0)
    if role == "submitter":
        return qs.filter(accepted_submission_count=0)
    return qs

filter_reachable(queryset)source

def filter_reachable(queryset):
    return queryset.filter(REACHABLE_SPEAKER_FILTER)

other_speaker_profiles(speaker)source

def other_speaker_profiles(speaker):
    with scopes_disabled():
        return SpeakerProfile.objects.filter(user_id=speaker.user_id).exclude(
            pk=speaker.pk
        )

speaker_by_email(event, email)source

def speaker_by_email(event, email):
    profiles = submitters_for_event(event, include_bare=True).filter(
        effective_email__iexact=email
    )
    profile = (
        profiles.annotate(
            has_submissions=Exists(SpeakerRole.objects.filter(speaker=OuterRef("pk")))
        )
        .order_by("-has_submissions", "pk")
        .first()
    )
    if profile:
        profile.event = event
    return profile

speakers_for_event(event)sourceSpeakers visible in event's current released schedule.

Follows talks_for_event to prevent diverging between public sessions and public speakers.

def speakers_for_event(event):
    """Speakers visible in ``event``'s current released schedule.

    Follows ``talks_for_event`` to prevent diverging between public
    sessions and public speakers.
    """
    return (
        SpeakerProfile.objects.filter(submissions__in=talks_for_event(event))
        .select_related("event", "user", "profile_picture")
        .order_by("id")
        .distinct()
    )

submitters_for_event(event, include_bare=False)sourceSpeakers who have any non-draft submission to event.

With include_bare, session-less speakers of non-CfP origin are included. CfP-origin speakers without any sessions are excluded as they are likely noise or have only drafts.

def submitters_for_event(event, include_bare=False):
    """Speakers who have any non-draft submission to ``event``.

    With include_bare, session-less speakers of non-CfP origin are
    included. CfP-origin speakers without any sessions are excluded
    as they are likely noise or have only drafts.
    """
    submitter_filter = Q(submissions__in=event.submissions.all())
    if include_bare:
        submitter_filter |= Q(
            submissions__isnull=True,
            origin__in=(SpeakerProfileOrigin.ORGA, SpeakerProfileOrigin.IMPORT),
        )
    return (
        SpeakerProfile.objects.filter(submitter_filter, event=event)
        .select_related("event", "user", "profile_picture")
        .order_by("id")
        .distinct()
    )

visible_talk_slots(speaker, schedule=None)sourceVisible talk slots for speaker in schedule (or the event's current schedule).

Returns an empty queryset when no schedule exists. Speakers within each slot are pre-sorted by their position.

def visible_talk_slots(speaker, schedule=None):
    """Visible talk slots for ``speaker`` in ``schedule`` (or the event's
    current schedule).

    Returns an empty queryset when no schedule exists. Speakers within each
    slot are pre-sorted by their position.
    """
    schedule = schedule or speaker.event.current_schedule
    if not schedule:
        return TalkSlot.objects.none()
    queryset = (
        schedule.talks.filter(submission__speakers=speaker, is_visible=True)
        .select_related(
            "submission",
            "room",
            "submission__event",
            "submission__track",
            "submission__submission_type",
        )
        .with_sorted_speakers()
    )
    if schedule.event.get_feature_flag("attendee_signup"):
        queryset = annotate_slot_signup_status(queryset)
    return queryset