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 onegov.pay 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
from typing import cast, Any, Literal, TYPE_CHECKING
if TYPE_CHECKING:
import uuid
# type gets shadowed by type in model, so we use Type as an alias
from builtins import type as type_t
from collections.abc 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 onegov.pay import Payment, PaymentError, PaymentProvider
from onegov.pay.types import PaymentMethod
from sqlalchemy.orm import Query
from typing 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 Scheduler.name, we have to pretend we
# created a subclass
class _OurScheduler(Scheduler):
name: uuid.UUID # type:ignore[assignment]
@lru_cache(maxsize=1)
[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 `<https://docs.sqlalchemy.org/en/improve_toc/
#: 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='Resource.id == Allocation.resource',
foreign_keys='Allocation.resource'
)
#: the date to jump to in the view (if not None) -> not in the db!
[docs]
date: datetime.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 = [a.id for a in allocations]
self.highlights_min = min(highlights)
self.highlights_max = max(highlights)
self.date = allocations[0].start.date()
[docs]
def get_scheduler(self, libres_context: 'Context') -> '_OurScheduler':
assert self.id, '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,
self.id, # type:ignore[arg-type]
self.timezone,
**extra_scheduler_arguments()
)
@property
[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
@property
[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, Price.zero())
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: datetime.date) -> bool:
if not self.zipcode_block:
return False
today = datetime.date.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}, {self.group}'