Source code for feriennet.models.group_invite

from functools import cached_property
from onegov.activity.models import Attendee, Booking, Occasion, Period
from onegov.activity.utils import random_group_code
from onegov.user import User
from sqlalchemy import func, or_
from sqlalchemy.orm import joinedload

from typing import TYPE_CHECKING
    from import Iterator
    from sqlalchemy.orm import Query, Session
    from typing_extensions import Self

[docs] class GroupInvite: def __init__( self, session: 'Session', group_code: str, username: str | None ) -> None: self.session = session self.group_code = group_code self.username = username @classmethod
[docs] def create(cls, session: 'Session', username: str | None) -> 'Self': """ Creates a new group invite with a code that is not yet used. """ candidate = cls( session=session, group_code=random_group_code(), username=username) # this might happen once in 26⁹ tries 🤞 if candidate.exists: return cls.create(session, username) return candidate
[docs] def for_username(self, username: str | None) -> 'Self': return self.__class__(self.session, self.group_code, username)
[docs] def user(self) -> User | None: if not self.username: return None return ( self.session.query(User) .filter_by(username=self.username).first() )
[docs] def exists(self) -> bool: """ Returns True if the group_code associated with this invite exists. """ return self.session.query(self.bookings().exists()).scalar()
[docs] def bookings(self) -> 'Query[Booking]': """ Returns a query of the bookings associated with this invite. """ return ( self.session.query(Booking) .options(joinedload(Booking.attendee)) .options(joinedload(Booking.occasion)) .options(joinedload(Booking.period)) .filter_by(group_code=self.group_code) .filter(or_( Booking.state.in_(('open', 'accepted')), Period.confirmed == False )) )
[docs] def occasion(self) -> Occasion: """ Looks up the occasion linked to this group invite. Technically it would be possible that a group code points to multiple occasions, but that would be an error. If that happens, an exception will be thrown. """ return self.session.query(Occasion).filter( self.bookings().with_entities(Booking.occasion_id).subquery() )).one()
[docs] def attendees(self) -> tuple[tuple[Attendee, Booking], ...]: """ Returns the attendees linked to this invite. """ return tuple( (booking.attendee, booking) for booking in self.bookings() # FIXME: Why is this an outerjoin? attendee_id is not nullable # so a regular join should work just fine .outerjoin(Attendee) .order_by(func.unaccent( )
[docs] def prospects( self, username: str ) -> 'Iterator[tuple[Attendee, Booking | None]]': """ Returns the attendees associated with the given users that are not yet part of the group. The result is a list of tuples with the first element being the attendee and the second element being the booking for the linked occasion, if such a booking already exists (otherwise None). """ if not username: return existing = { for a, b in self.attendees} attendees = ( self.session.query(Attendee) .filter(Attendee.username == username) .order_by(func.unaccent( ) bookings_query = ( self.session.query(Booking) .filter(Booking.occasion_id == .filter(Booking.attendee_id.in_( attendees.with_entities( ) bookings = {b.attendee_id: b for b in bookings_query} for attendee in attendees: if not in existing: yield attendee, bookings.get(, None)