Source code for activity.models.period

import sedate

from datetime import date, datetime
from onegov.activity.models.age_barrier import AgeBarrier
from onegov.activity.models.booking import Booking
from onegov.activity.models.occasion import Occasion
from onegov.core.orm import Base
from onegov.core.orm.mixins import TimestampMixin
from onegov.core.orm.types import UUID, JSON
from sqlalchemy import Boolean
from sqlalchemy import desc, not_, distinct
from sqlalchemy import CheckConstraint
from sqlalchemy import column
from sqlalchemy import Column
from sqlalchemy import Date
from sqlalchemy import Index
from sqlalchemy import Integer
from sqlalchemy import Numeric
from sqlalchemy import Text
from sqlalchemy.orm import object_session, relationship, joinedload, defer
from sqlalchemy.orm import validates
from uuid import uuid4


from typing import Any, ClassVar, NamedTuple, NoReturn, TYPE_CHECKING
if TYPE_CHECKING:
    import uuid
    from collections.abc import Iterator
    from decimal import Decimal
    from onegov.activity.matching.score import Scoring
    from onegov.activity.models import Invoice, PublicationRequest
    from sqlalchemy.orm import Session


[docs] class PeriodMixin: # It's doubtful that the Ferienpass would ever run anywhere else but # in Switzerland ;)
[docs] timezone: ClassVar[str] = 'Europe/Zurich'
if TYPE_CHECKING: # forward declare required attributes @property
[docs] def active(self) -> Column[bool] | bool: ...
@property def confirmed(self) -> Column[bool] | bool: ... @property def finalized(self) -> Column[bool] | bool: ... @property def prebooking_start(self) -> Column[date] | date: ... @property def prebooking_end(self) -> Column[date] | date: ... @property def booking_start(self) -> Column[date] | date: ... @property def booking_end(self) -> Column[date] | date: ... @property def execution_start(self) -> Column[date] | date: ... @property def execution_end(self) -> Column[date] | date: ... @property def book_finalized(self) -> Column[bool] | bool: ... @property def max_bookings_per_attendee( self ) -> Column[int | None] | int | None: ...
[docs] def as_local_datetime( self, day: date | datetime, end_of_day: bool = False ) -> datetime: """ Returns the moment of midnight in terms of the timezone it UTC """ return sedate.standardize_date( datetime( day.year, day.month, day.day, 23 if end_of_day else 0, 59 if end_of_day else 0, 59 if end_of_day else 0 ), self.timezone )
@property
[docs] def phase(self) -> str | None: local = self.as_local_datetime now = sedate.utcnow() if not self.active or now < local(self.prebooking_start): return 'inactive' if not self.confirmed: return 'wishlist' if now < local(self.booking_start): return 'inactive' if not self.finalized and local(self.booking_end, True) < now: return 'inactive' if not self.finalized: return 'booking' local_execution_start = local(self.execution_start) if now < local_execution_start: return 'payment' local_execution_end = local(self.execution_end, end_of_day=True) if local_execution_start <= now <= local_execution_end: return 'execution' if now > local_execution_end: return 'archive' # FIXME: Is this allowed? return None
@property
[docs] def wishlist_phase(self) -> bool: return self.phase == 'wishlist'
@property
[docs] def booking_phase(self) -> bool: return self.phase == 'booking'
@property
[docs] def payment_phase(self) -> bool: return self.phase == 'payment'
@property
[docs] def execution_phase(self) -> bool: return self.phase == 'execution'
@property
[docs] def archive_phase(self) -> bool: return self.phase == 'archive'
@property
[docs] def is_prebooking_in_future(self) -> bool: now = sedate.utcnow() start = self.as_local_datetime(self.prebooking_start) return now < start
@property
[docs] def is_currently_prebooking(self) -> bool: if not self.wishlist_phase: return False now = sedate.utcnow() start = self.as_local_datetime(self.prebooking_start) end = self.as_local_datetime(self.prebooking_end, end_of_day=True) return start <= now <= end
@property
[docs] def is_prebooking_in_past(self) -> bool: """Returns true if current date is after start of booking phase or if current date is after prebooking end. """ now = sedate.utcnow() start = self.as_local_datetime(self.prebooking_start) end = self.as_local_datetime(self.prebooking_end, end_of_day=True) if now > end: return True return start <= now and not self.wishlist_phase
@property
[docs] def is_booking_in_future(self) -> bool: now = sedate.utcnow() start = self.as_local_datetime(self.booking_start) return now < start
@property
[docs] def is_currently_booking(self) -> bool: if not self.booking_phase: return False now = sedate.utcnow() start = self.as_local_datetime(self.booking_start) end = self.as_local_datetime(self.booking_end, end_of_day=True) return start <= now <= end
@property
[docs] def is_booking_in_past(self) -> bool: now = sedate.utcnow() start = self.as_local_datetime(self.booking_start) end = self.as_local_datetime(self.booking_end, end_of_day=True) if now > end: return True return start <= now and not ( self.booking_phase or self.book_finalized)
@property
[docs] def is_execution_in_past(self) -> bool: now = sedate.utcnow() end = self.as_local_datetime(self.execution_end, end_of_day=True) return now > end
@property
[docs] def booking_limit(self) -> int | None: """ Returns the max_bookings_per_attendee limit if it applies. """ return self.max_bookings_per_attendee
[docs] class PeriodMetaBase(NamedTuple):
[docs] id: 'uuid.UUID'
[docs] title: str
[docs] active: bool
[docs] confirmed: bool
[docs] confirmable: bool
[docs] finalized: bool
[docs] finalizable: bool
[docs] archived: bool
[docs] prebooking_start: date
[docs] prebooking_end: date
[docs] booking_start: date
[docs] booking_end: date
[docs] execution_start: date
[docs] execution_end: date
[docs] max_bookings_per_attendee: int | None
[docs] booking_cost: 'Decimal | None'
[docs] all_inclusive: bool
[docs] pay_organiser_directly: bool
[docs] minutes_between: int | None
[docs] alignment: str | None
[docs] deadline_days: int | None
[docs] book_finalized: bool
[docs] cancellation_date: date | None
[docs] cancellation_days: int | None
[docs] age_barrier_type: str
[docs] class PeriodMeta(PeriodMetaBase, PeriodMixin): # TODO: We would like to add a request scoped cache to # all the properties on PeriodMixin, since they would # likely not change within the time span of a single # request, but we can't use `cached_property`, since # that would risk staying around across multiple requests # alternatively we could add a time-based cache with # a short TTL like 60 seconds. That would already avoid # the many redundant calls to `phase`.
[docs] def materialize(self, session: 'Session') -> 'Period': period = session.query(Period).get(self.id) assert period is not None return period
[docs] def __setattr__(self, name: str, value: object) -> NoReturn: # NOTE: This is a guard against people setting attributes # that exist on Period but not on PeriodMeta and # getting silently dropped because of it. raise ValueError( 'You are not allowed to set attributes on PeriodMeta.' )
[docs] class Period(Base, PeriodMixin, TimestampMixin):
[docs] __tablename__ = 'periods'
#: The public id of this period
[docs] id: 'Column[uuid.UUID]' = Column( UUID, # type:ignore[arg-type] primary_key=True, default=uuid4 )
#: The public title of this period
[docs] title: 'Column[str]' = Column(Text, nullable=False)
#: Only one period is active at a time
[docs] active: 'Column[bool]' = Column(Boolean, nullable=False, default=False)
#: A confirmed period may not be automatically matched anymore and all #: booking changes to it are communicated to the customer
[docs] confirmed: 'Column[bool]' = Column(Boolean, nullable=False, default=False)
#: A confirmable period has a prebooking phase, while an unconfirmable # booking does not. An unconfirmable booking starts as `confirmed` for # legacy reasons (even though it doesn't sound sane to have an # unconfirmable period that is confirmed).
[docs] confirmable: 'Column[bool]' = Column(Boolean, nullable=False, default=True)
#: A finalized period may not have any change in bookings anymore
[docs] finalized: 'Column[bool]' = Column(Boolean, nullable=False, default=False)
#: A finalizable period may have invoices associated with it, an #: unfinalizable period may not
[docs] finalizable: 'Column[bool]' = Column(Boolean, nullable=False, default=True)
#: An archived period has been entirely completed
[docs] archived: 'Column[bool]' = Column(Boolean, nullable=False, default=False)
#: Start of the wishlist-phase
[docs] prebooking_start: 'Column[date]' = Column(Date, nullable=False)
#: End of the wishlist-phase
[docs] prebooking_end: 'Column[date]' = Column(Date, nullable=False)
#: Start of the booking-phase
[docs] booking_start: 'Column[date]' = Column(Date, nullable=False)
#: End of the booking-phase
[docs] booking_end: 'Column[date]' = Column(Date, nullable=False)
#: Date of the earliest possible occasion start of this period
[docs] execution_start: 'Column[date]' = Column(Date, nullable=False)
#: Date of the latest possible occasion end of this period
[docs] execution_end: 'Column[date]' = Column(Date, nullable=False)
#: Extra data stored on the period
[docs] data: 'Column[dict[str, Any]]' = Column(JSON, nullable=False, default=dict)
#: Maximum number of bookings per attendee
[docs] max_bookings_per_attendee: 'Column[int | None]' = Column( Integer, nullable=True )
#: Base cost for one or many bookings
[docs] booking_cost: 'Column[Decimal | None]' = Column( Numeric(precision=8, scale=2), nullable=True )
#: True if the booking cost is meant for all bookings in a period #: or for each single booking
[docs] all_inclusive: 'Column[bool]' = Column( Boolean, nullable=False, default=False )
#: True if the costs of an occasions need to be paid to the organiser
[docs] pay_organiser_directly: 'Column[bool]' = Column( Boolean, nullable=False, default=False )
#: Time between bookings in minutes
[docs] minutes_between: 'Column[int | None]' = Column( Integer, nullable=True, default=0 )
#: The alignment of bookings in the matching # FIXME: Restrict this to what is actually allowed i.e. Literal['day', ...]
[docs] alignment: 'Column[str | None]' = Column(Text, nullable=True)
#: Deadline for booking occasions. A deadline of 3 means that 3 days before #: an occasion is set to start, bookings are disabled. #: #: Note, unless book_finalized is set to True, this setting has no effect #: in a finalized period. #: #: Also, if deadline_days is None, bookings can't be created in a #: finalized period either, as deadline_days is a prerequisite for the #: book_finalized setting.
[docs] deadline_days: 'Column[int | None]' = Column(Integer, nullable=True)
#: True if bookings can be created by normal users in finalized periods. #: The deadline_days are still applied for these normal users. #: Admins can always create bookings during any time, deadline_days and #: book_finalized are ignored.
[docs] book_finalized: 'Column[bool]' = Column( Boolean, nullable=False, default=False )
#: Date after which no bookings can be canceled by a mere member
[docs] cancellation_date: 'Column[date | None]' = Column(Date, nullable=True)
#: Days between the occasion and the cancellation (an alternative to #: the cancellation_date)
[docs] cancellation_days: 'Column[int | None]' = Column(Integer, nullable=True)
#: The age barrier implementation in use
[docs] age_barrier_type: 'Column[str]' = Column( Text, nullable=False, default='exact' )
[docs] __table_args__ = ( CheckConstraint(( # ranges should be valid 'prebooking_start <= prebooking_end AND ' 'booking_start <= booking_end AND ' 'execution_start <= execution_end AND ' # pre-booking must happen before booking and execution 'prebooking_end <= booking_start AND ' 'prebooking_end <= execution_start AND ' # booking and execution may overlap, but the execution cannot # start before booking begins 'booking_start <= execution_start AND ' 'booking_end <= execution_end' ), name='period_date_order'), Index( 'only_one_active_period', 'active', unique=True, postgresql_where=column('active') == True ) )
#: The occasions linked to this period
[docs] occasions: 'relationship[list[Occasion]]' = relationship( 'Occasion', order_by='Occasion.order', back_populates='period' )
#: The bookings linked to this period
[docs] bookings: 'relationship[list[Booking]]' = relationship( 'Booking', back_populates='period' )
[docs] invoices: 'relationship[list[Invoice]]' = relationship( 'Invoice', back_populates='period' )
[docs] publication_requests: 'relationship[list[PublicationRequest]]'
publication_requests = relationship( 'PublicationRequest', back_populates='period' ) @validates('age_barrier_type')
[docs] def validate_age_barrier_type( self, key: str, age_barrier_type: str ) -> str: assert age_barrier_type in AgeBarrier.registry return age_barrier_type
@property
[docs] def age_barrier(self) -> AgeBarrier: return AgeBarrier.from_name(self.age_barrier_type)
[docs] def activate(self) -> None: """ Activates the current period, causing all occasions and activites to update their status and book-keeping. It also makes sure no other period is active. """ if self.active: return session = object_session(self) model = self.__class__ active_period = ( session.query(model) .filter(model.active == True).first() ) if active_period: active_period.deactivate() # avoid triggering the only_one_active_period index constraint session.flush() self.active = True
[docs] def deactivate(self) -> None: """ Deactivates the current period, causing all occasions and activites to update their status and book-keeping. """ if not self.active: return self.active = False
[docs] def confirm(self) -> None: """ Confirms the current period. """ self.confirmed = True # open bookings are marked as denied during completion # and the booking costs are copied over permanently (so they can't # change anymore) b = object_session(self).query(Booking) b = b.filter(Booking.period_id == self.id) b = b.options(joinedload(Booking.occasion)) b = b.options( defer(Booking.group_code), defer(Booking.attendee_id), defer(Booking.priority), defer(Booking.username), ) for booking in b: if booking.state == 'open': booking.state = 'denied' booking.cost = booking.occasion.total_cost
[docs] def archive(self) -> None: """ Moves all accepted activities with an occasion in this period into the archived state, unless there's already another occasion in a period newer than the current period. """ assert self.confirmed and self.finalized or not self.finalizable self.archived = True self.active = False session = object_session(self) def future_periods() -> 'Iterator[uuid.UUID]': p = session.query(Period) p = p.order_by(desc(Period.execution_start)) p = p.with_entities(Period.id) for period in p: if period.id == self.id: break yield period.id # get the activities which have an occasion in a future period f = session.query(Occasion) f = f.with_entities(Occasion.activity_id) f = f.filter(Occasion.period_id.in_(tuple(future_periods()))) # get the activities which have an occasion in the given period but # no occasion in any future period o = session.query(Occasion) o = o.filter(Occasion.period_id == self.id) o = o.filter(not_(Occasion.activity_id.in_(f.subquery()))) o = o.options(joinedload(Occasion.activity)) # archive those for occasion in o: if occasion.activity.state == 'accepted': occasion.activity.archive() # also archive all activities without an occasion w = session.query(Occasion) w = w.with_entities(distinct(Occasion.activity_id)) # XXX circular import from onegov.activity.models.activity import Activity a = session.query(Activity) a = a.filter(not_(Activity.id.in_(w.subquery()))) a = a.filter(Activity.state == 'accepted') for activity in a: activity.archive()
[docs] def confirm_and_start_booking_phase(self) -> None: """ Confirms the period and sets the booking phase to now. This is mainly an internal convenience function to activate the previous behaviour before a specific booking phase date was introduced. """ self.confirmed = True self.prebooking_end = date.today() self.booking_start = date.today()
@property
[docs] def scoring(self) -> 'Scoring': # circular import from onegov.activity.matching.score import Scoring return Scoring.from_settings( settings=self.data.get('match-settings', {}), session=object_session(self))
@scoring.setter def scoring(self, scoring: 'Scoring') -> None: self.data['match-settings'] = scoring.settings
[docs] def materialize(self, session: 'Session') -> 'Period': return self