Source code for form.models.registration_window

import sedate

from onegov.core.orm import Base
from onegov.core.orm.mixins import TimestampMixin
from onegov.core.orm.types import UUID
from onegov.form.models.submission import FormSubmission
from sqlalchemy import and_
from sqlalchemy import Boolean
from sqlalchemy import Column
from sqlalchemy import Date
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import or_
from sqlalchemy import Text
from sqlalchemy import text
from sqlalchemy.dialects.postgresql import ExcludeConstraint
from sqlalchemy.orm import object_session, relationship
from sqlalchemy.schema import CheckConstraint
from sqlalchemy.sql.elements import quoted_name
from uuid import uuid4

from typing import TYPE_CHECKING
    import uuid
    from datetime import date, datetime
    from onegov.form.models.definition import FormDefinition

[docs] daterange = Column( # type:ignore[call-overload] quoted_name('DATERANGE("start", "end")', quote=False))
[docs] class FormRegistrationWindow(Base, TimestampMixin): """ Defines a registration window during which a form definition may be used to create submissions. Submissions created thusly are attached to the currently active registration window. Registration windows may not overlap. """
[docs] __tablename__ = 'registration_windows'
#: the public id of the registraiton window
[docs] id: 'Column[uuid.UUID]' = Column( UUID, # type:ignore[arg-type] primary_key=True, default=uuid4 )
#: the name of the form to which this registration window belongs
[docs] name: 'Column[str]' = Column( Text, ForeignKey(""), nullable=False )
#: true if the registration window is enabled
[docs] enabled: 'Column[bool]' = Column(Boolean, nullable=False, default=True)
#: the start date of the window
[docs] start: 'Column[date]' = Column(Date, nullable=False)
#: the end date of the window
[docs] end: 'Column[date]' = Column(Date, nullable=False)
#: the timezone of the window
[docs] timezone: 'Column[str]' = Column( Text, nullable=False, default='Europe/Zurich' )
#: the number of spots (None => unlimited)
[docs] limit: 'Column[int | None]' = Column(Integer, nullable=True)
#: enable an overflow of submissions
[docs] overflow: 'Column[bool]' = Column(Boolean, nullable=False, default=True)
#: submissions linked to this
[docs] submissions: 'relationship[list[FormSubmission]]' = relationship( FormSubmission, backref='registration_window' )
if TYPE_CHECKING: # forward declare backref form: relationship[FormDefinition]
[docs] __table_args__ = ( # ensures that there are no overlapping date ranges within one form ExcludeConstraint( (name, '='), (daterange, '&&'), name='no_overlapping_registration_windows', using='gist' ), # ensures that there are no adjacent date ranges # (end on the same day as next start) ExcludeConstraint( (name, '='), (daterange, '-|-'), name='no_adjacent_registration_windows', using='gist' ), # ensures that start <= end CheckConstraint( '"start" <= "end"', name='start_smaller_than_end' ), )
[docs] def localized_start(self) -> 'datetime': return sedate.align_date_to_day( sedate.standardize_date( sedate.as_datetime(self.start), self.timezone ), self.timezone, 'down' )
[docs] def localized_end(self) -> 'datetime': return sedate.align_date_to_day( sedate.standardize_date( sedate.as_datetime(self.end), self.timezone ), self.timezone, 'up' )
[docs] def disassociate(self) -> None: """ Disassociates all records linked to this window. """ for submission in self.submissions: submission.disclaim() submission.registration_window_id = None
[docs] def in_the_future(self) -> bool: return sedate.utcnow() <= self.localized_start
[docs] def in_the_past(self) -> bool: return self.localized_end <= sedate.utcnow()
[docs] def in_the_present(self) -> bool: return self.localized_start <= sedate.utcnow() <= self.localized_end
[docs] def accepts_submissions(self, required_spots: int = 1) -> bool: assert required_spots > 0 if not self.enabled: return False if not self.in_the_present: return False if self.overflow: return True if self.limit is None: return True return self.available_spots >= required_spots
[docs] def next_submission(self) -> FormSubmission | None: """ Returns the submission next in line. In other words, the next submission in order of first come, first serve. """ q = object_session(self).query(FormSubmission) q = q.filter(FormSubmission.registration_window_id == q = q.filter(FormSubmission.state == 'complete') q = q.filter(or_( FormSubmission.claimed == None, and_( FormSubmission.claimed > 0, FormSubmission.claimed < FormSubmission.spots, ) )) q = q.order_by(FormSubmission.created) return q.first()
[docs] def available_spots(self) -> int: assert self.limit is not None return max(self.limit - self.claimed_spots - self.requested_spots, 0)
[docs] def claimed_spots(self) -> int: """ Returns the number of actually claimed spots. """ return object_session(self).execute(text(""" SELECT SUM(COALESCE(claimed, 0)) FROM submissions WHERE registration_window_id = :id AND submissions.state = 'complete' """), {'id':}).scalar() or 0
[docs] def requested_spots(self) -> int: """ Returns the number of requested spots. When the claim has not been made yet, `spots` are counted as requested. When the claim has been partially made, the difference is counted as requested. If the claim has been fully made, the result is 0. If the claim has been relinquished, the result is 0. """ return object_session(self).execute(text(""" SELECT GREATEST( SUM( CASE WHEN claimed IS NULL THEN spots WHEN claimed = 0 THEN 0 ELSE spots - claimed END ), 0 ) FROM submissions WHERE registration_window_id = :id AND submissions.state = 'complete' """), {'id':}).scalar() or 0