Source code for reservation.models.resource

import datetime
import secrets

from libres import new_scheduler
from libres.db.models import Allocation
from libres.db.models.base import ORMBase
from onegov.core.cache import lru_cache
from onegov.core.orm import ModelBase
from onegov.core.orm.mixins import content_property, dict_property
from onegov.core.orm.mixins import ContentMixin, TimestampMixin
from onegov.core.orm.types import UUID
from onegov.file import MultiAssociatedFiles
from onegov.form import parse_form
from import Price, process_payment
from sedate import align_date_to_day, utcnow
from sqlalchemy import Column, Text
from sqlalchemy.orm import relationship
from uuid import uuid4

# type gets shadowed by type in model, so we use Type as an alias
from typing import cast, Any, Literal, Type, TYPE_CHECKING
    import uuid
    from import Sequence
    from libres.context.core import Context
    from libres.db.scheduler import Scheduler
    from onegov.form import Form
    from onegov.reservation.models import CustomReservation
    from import Payment, PaymentError, PaymentProvider
    from import PaymentMethod
    from sqlalchemy.orm import Query
    from typing_extensions import TypeAlias

[docs] DeadlineUnit: TypeAlias = Literal['d', 'h']
# HACK: We pass a UUID as a name and have a custom uuid_generator # which directly uses it, so in order to get the correct # type checking on, we have to pretend we # created a subclass class _OurScheduler(Scheduler): name: uuid.UUID # type:ignore[assignment] @lru_cache(maxsize=1)
[docs] def extra_scheduler_arguments() -> dict[str, Any]: from onegov.reservation.models import CustomReservation from onegov.reservation.models import CustomAllocation return { 'allocation_cls': CustomAllocation, 'reservation_cls': CustomReservation }
[docs] class Resource(ORMBase, ModelBase, ContentMixin, TimestampMixin, MultiAssociatedFiles): """ A resource holds a single calendar with allocations and reservations. Note that this resource is not defined on the onegov.core declarative base. Instead it is defined using the libres base. This means we can't join data outside the libres models. This should however not be a problem as this onegov module is self contained and does not link to other onegov modules, except for core. If we ever want to link to other models (say link a reservation to a user), then we have to switch to a unified base. Ideally we would find a way to merge these bases somehow. Also note that we *do* use the ModelBase class as a mixin to at least share the same methods as all the usual onegov.core.orm models. """
[docs] __tablename__ = 'resources'
#: the unique id
[docs] id: 'Column[uuid.UUID]' = Column( UUID, # type:ignore[arg-type] primary_key=True, default=uuid4 )
#: a nice id for the url, readable by humans # FIXME: This probably should've been nullable=False
[docs] name: 'Column[str | None]' = Column(Text, primary_key=False, unique=True)
#: the title of the resource
[docs] title: 'Column[str]' = Column(Text, primary_key=False, nullable=False)
#: the timezone this resource resides in
[docs] timezone: 'Column[str]' = Column(Text, nullable=False)
#: the custom form definition used when creating a reservation
[docs] definition: 'Column[str | None]' = Column(Text, nullable=True)
#: the group to which this resource belongs to (may be any kind of string)
[docs] group: 'Column[str | None]' = Column(Text, nullable=True)
#: the type of the resource, this can be used to create custom polymorphic #: subclasses. See `< #: orm/extensions/declarative/inheritance.html>`_.
[docs] type: 'Column[str]' = Column( Text, nullable=False, default=lambda: 'generic' )
#: the payment method
[docs] payment_method: dict_property['PaymentMethod | None'] = content_property()
#: the minimum price total the reservation must exceed
[docs] minimum_price_total: dict_property[float | None] = content_property()
#: the currency of the price to pay
[docs] currency: dict_property[str | None] = content_property()
#: the pricing method to use
[docs] pricing_method: dict_property[str | None] = content_property()
#: the reservations cost a given amount per hour
[docs] price_per_hour: dict_property[float | None] = content_property()
#: the reservations cost a given amount per unit (allocations * quota)
[docs] price_per_item: dict_property[float | None]
price_per_item = content_property('price_per_reservation') #: reservation deadline (e.g. None, (5, 'd'), (24, 'h'))
[docs] deadline: dict_property[tuple[int, 'DeadlineUnit'] | None]
deadline = content_property() #: the default view
[docs] default_view: dict_property[str | None] = content_property()
#: reservation zip code limit, contains None or something like this: #: { #: 'zipcode_field': 'PLZ', #: 'zipcode_list': [1234, 5678], #: 'zipcode_days': 3 #: } #: #: zipcode_field -> the field name in the definition containing zip codes #: zipcode_list -> zip codes exempt from the rule #: zipcode_days -> how many days before the reservation the rule is dropped #: #: Note, the zipcode_field name is in the human readable form. # FIXME: Define a TypedDict with all the zipcode_block elements
[docs] zipcode_block: dict_property[dict[str, Any] | None] = content_property()
#: secret token to get anonymous access to calendar data
[docs] access_token: dict_property[str | None] = content_property()
#: hint on how to get to the resource
[docs] pick_up: dict_property[str | None] = content_property()
[docs] __mapper_args__ = { "polymorphic_on": 'type', 'polymorphic_identity': 'generic' }
[docs] allocations: 'relationship[list[Allocation]]' = relationship( Allocation, cascade="all, delete-orphan", primaryjoin=' == Allocation.resource', foreign_keys='Allocation.resource' )
#: the date to jump to in the view (if not None) -> not in the db!
[docs] date: | None = None
#: a range of allocation ids to highlight in the view (if not None)
[docs] highlights_min: int | None = None
[docs] highlights_max: int | None = None
#: the view to open in the calendar (fullCalendar view name)
[docs] view: str | None = 'month'
@deadline.setter def set_deadline(self, value: tuple[int, 'DeadlineUnit'] | None) -> None: value = value or None if value: if len(value) != 2: raise ValueError("Deadline is not a tuple with two elements") if not isinstance(value[0], int): raise ValueError("Deadline value is not an int") if value[0] < 1: raise ValueError("Deadline value is smaller than 1") if value[1] not in ('d', 'h'): raise ValueError("Deadline unit must be 'd' or 'h'") self.content['deadline'] = value
[docs] def highlight_allocations( self, allocations: 'Sequence[Allocation]' ) -> None: """ The allocation to jump to in the view. """ # we can assume that allocation ids are created in a continuous # number line. It's not necessarily guaranteed, but since it *is* # only a highlighting feature we can check the highlights more # effiecently if we follow this assumption. highlights = [ for a in allocations] self.highlights_min = min(highlights) self.highlights_max = max(highlights) = allocations[0]
[docs] def get_scheduler(self, libres_context: 'Context') -> '_OurScheduler': assert, "the id needs to be set" assert self.timezone, "the timezone needs to be set" # HACK: we work around the name being a str in libres, but a # UUID in onegov return new_scheduler( # type:ignore[return-value] libres_context,, # type:ignore[arg-type] self.timezone, **extra_scheduler_arguments() )
[docs] def scheduler(self) -> '_OurScheduler': assert hasattr(self, 'libres_context'), "not bound to libres context" return self.get_scheduler(self.libres_context)
[docs] def bind_to_libres_context(self, libres_context: 'Context') -> None: self.libres_context = libres_context
[docs] def form_class(self) -> 'Type[Form] | None': """ Parses the form definition and returns a form class. """ if not self.definition: return None return parse_form(self.definition)
[docs] def price_of_reservation( self, token: 'uuid.UUID', extra: Price | None = None ) -> Price: # FIXME: libres is very laissez faire with the polymorphic # classes and always uses the base classes for queries # rather than the ones supplied to the Scheduler, so # we can't actually assume we get our Reservation class # unless we only ever create instances of our own class # inside the current context, this is not really acceptable # for type checking. We could pretend that the Scheduler # always gives us the class we bound do it, but that's # not technically true... _reservations = self.scheduler.reservations_by_token(token) reservations = cast('Query[CustomReservation]', _reservations) prices = (price for r in reservations if (price := r.price(self))) total = sum(prices, if extra and total: total += extra elif extra: total = extra return total
[docs] def process_payment( self, price: Price | None, provider: 'PaymentProvider[Any] | None' = None, payment_token: str | None = None ) -> 'Payment | PaymentError | Literal[True] | None': """ Processes the payment for the given reservation token. """ if price and price.amount > 0: assert self.payment_method is not None return process_payment( self.payment_method, price, provider, payment_token) # FIXME: Returning a boolean is a bit strange here, do we # make use of it or can we change this to None? return True
[docs] def is_past_deadline(self, dt: datetime.datetime) -> bool: if not self.deadline: return False if not dt.tzinfo: raise RuntimeError(f"The given date has no timezone: {dt}") if not self.timezone: raise RuntimeError("No timezone set on the resource") n, unit = self.deadline # hours result in a simple offset def deadline_using_h() -> datetime.datetime: return dt - datetime.timedelta(hours=n) # days require that we align the date to the beginning of the date def deadline_using_d() -> datetime.datetime: return ( align_date_to_day(dt, self.timezone, 'down') - datetime.timedelta(days=(n - 1)) ) deadline = locals()[f'deadline_using_{unit}']() return deadline <= utcnow()
[docs] def is_zip_blocked(self, date: -> bool: if not self.zipcode_block: return False today = return (date - today).days > self.zipcode_block['zipcode_days']
[docs] def is_allowed_zip_code(self, zipcode: int) -> bool: assert isinstance(zipcode, int) if not self.zipcode_block: return True return zipcode in self.zipcode_block['zipcode_list']
[docs] def renew_access_token(self) -> None: self.access_token = secrets.token_hex(16)
[docs] def __repr__(self) -> str: return f"{self.title}, {}"