import sedate
from datetime import date, datetime
from libres.db.models import ReservedSlot
from onegov.core.orm.mixins import (
dict_markup_property, dict_property, meta_property)
from onegov.form.models import FormSubmission
from onegov.org.i18n import _
from onegov.org.models.extensions import (
ContactExtension, GeneralFileLinkExtension,
InlinePhotoAlbumExtension, ResourceValidationExtension)
from onegov.org.models.extensions import CoordinatesExtension
from onegov.org.models.extensions import AccessExtension
from onegov.org.models.extensions import PersonLinkExtension
from onegov.reservation import Resource, ResourceCollection, Reservation
from onegov.search import SearchableContent
from onegov.ticket import Ticket
from sqlalchemy import UUID as UUIDType
from sqlalchemy.orm import undefer
from sqlalchemy.sql.expression import cast
from uuid import uuid4, uuid5, UUID
from typing import ClassVar, TYPE_CHECKING
if TYPE_CHECKING:
from libres.context.core import Context
from libres.db.scheduler import Scheduler
from onegov.org.request import OrgRequest
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import Mapped, Query
[docs]
class FindYourSpotCollection(ResourceCollection):
def __init__(
self,
libres_context: Context,
group: str | None = None,
subgroup: str | None = None,
) -> None:
super().__init__(libres_context)
[docs]
self.subgroup = subgroup
[docs]
def query(self) -> Query[Resource]:
query = self.session.query(Resource)
# we only support find-your-spot for rooms for now
query = query.filter(Resource.type == 'room')
query = query.filter(Resource.group == (self.group or ''))
if self.subgroup is not None:
query = query.filter(Resource.subgroup == self.subgroup)
return query
[docs]
class SharedMethods:
if TYPE_CHECKING:
[docs]
title_template: ClassVar[str]
id: Mapped[UUID]
libres_context: Context
date: date | None
view: str | None
timezone: Mapped[str]
virtual_id: hybrid_property[UUID]
@property
def scheduler(self) -> Scheduler: ...
@property
def virtual_scheduler(self) -> Scheduler: ...
def get_scheduler(
self,
context: Context,
*,
virtual: bool = False
) -> Scheduler: ...
[docs]
lead: dict_property[str | None] = meta_property()
[docs]
text = dict_markup_property('content')
[docs]
confirmation_text = dict_markup_property('content')
[docs]
occupancy_is_visible_to_members: dict_property[bool | None] = (
meta_property()
)
@property
[docs]
def deletable(self) -> bool:
# FIXME: use exists() subqueries for speed
if self.scheduler.managed_reserved_slots().first():
return False
if self.scheduler.managed_reservations().first():
return False
return True
@property
[docs]
def calendar_date_range(self) -> tuple[datetime, datetime]:
""" The date range set by the fullcalendar specific params. """
if self.date:
date = datetime(self.date.year, self.date.month, self.date.day)
date = sedate.replace_timezone(date, self.timezone)
else:
date = sedate.to_timezone(sedate.utcnow(), self.timezone)
if self.view in ('multiMonthYear', 'listYear'):
return sedate.replace_timezone(
datetime(date.year, 1, 1),
self.timezone
), sedate.replace_timezone(
datetime(date.year, 12, 31, 23, 59, 59, 999999),
self.timezone
)
elif self.view in ('dayGridMonth', 'listMonth'):
return sedate.align_range_to_month(date, date, self.timezone)
elif self.view in ('timeGridWeek', 'listWeek'):
return sedate.align_range_to_week(date, date, self.timezone)
elif self.view in ('timeGridDay', 'listDay'):
return sedate.align_range_to_day(date, date, self.timezone)
else:
raise NotImplementedError()
[docs]
def remove_expired_reservation_sessions(
self,
expiration_date: datetime | None = None
) -> None:
# FIXME: Ideally this would also clean up any virtual allocations
# that are no longer needed, but since this clean up isn't
# scoped to a specific resource, we'd have to manually write
# it or incur an unreasonably high cost, for something that
# is intended to run at the start of some requests. If we
# refactor this into a cronjob instead, we could perhaps
# get away with doing something more expensive. But we also
# might just want to add a cheaper bulk query to libres that
# does this for us or write the query here and make sure it's
# well tested.
session = self.libres_context.get_service('session_provider').session()
queries = self.scheduler.queries
expired_sessions = queries.find_expired_reservation_sessions(
expiration_date)
if expired_sessions:
query = session.query(Reservation).with_entities(Reservation.token)
query = query.filter(Reservation.session_id.in_(expired_sessions))
tokens = {token for token, in query}
query = session.query(FormSubmission)
query = query.filter(FormSubmission.name == None)
query = query.filter(FormSubmission.id.in_(tokens))
# NOTE: This used to be a batch delete, but since there may be
# files attached to these submissions, we would need to
# emit a batch delete for the file links and any orphaned
# files. For now it's easier to do single deletes so
# SQLAlchemy handles this for us, most of the time this
# will only hit one or two submissions anyways. There
# would need to be a burst of submissions where everyone
# got as far as entering all of their data, but didn't
# end up confirming the reservation and then a long
# period after where no reservations happen at all.
for submission in query:
# remove file links (also removes orphaned files)
submission.files = []
session.delete(submission)
queries.remove_expired_reservation_sessions(expiration_date)
[docs]
def bound_reservations(
self,
request: OrgRequest,
status: str = 'pending',
with_data: bool = False,
with_virtual: bool = False
) -> Query[Reservation]:
""" The reservations associated with this resource and user. """
session = self.bound_session_id(request)
scheduler = self.get_scheduler(request.app.libres_context)
res = scheduler.queries.reservations_by_session(session)
if with_virtual:
res = res.filter(Reservation.resource.in_(
[self.id, self.virtual_id]
))
else:
res = res.filter(Reservation.resource == self.id)
res = res.filter(Reservation.status == status)
res = res.order_by(None) # clear existing order
res = res.order_by(Reservation.start)
# used by ReservationInfo
res = res.options(undefer(Reservation.created))
if with_data:
res = res.options(undefer(Reservation.data))
return res # type: ignore[return-value]
[docs]
def bound_session_id(self, request: OrgRequest) -> UUID:
""" The session id associated with this resource and user. """
if not request.browser_session.has('libres_session_id'):
request.browser_session.libres_session_id = uuid4()
return uuid5(self.id, request.browser_session.libres_session_id.hex)
[docs]
def reservations_with_tickets_query(
self,
start: datetime | None = None,
end: datetime | None = None,
exclude_pending: bool = True,
only_managed: bool = True,
# FIXME: This kind of conflicts with only_managed, maybe we should
# merge these two boolean arguments into a ternary argument
# indicating which reservations you want:
# only_managed, include_virtual, all_visible
include_virtual: bool = False
) -> Query[Reservation]:
""" Returns a query which joins this resource's reservations between
start and end with the tickets table.
"""
query = self.scheduler.session.query(Reservation)
resource_ids = {self.id}
if not only_managed:
resource_ids.update(self.scheduler.blocking_resources)
elif include_virtual:
resource_ids.add(self.virtual_id)
query = query.filter(Reservation.resource.in_(resource_ids))
if start:
query = query.filter(start <= Reservation.start)
if end:
query = query.filter(Reservation.end <= end)
query = query.join(
Ticket, Reservation.token == cast(Ticket.handler_id, UUIDType))
query = query.order_by(Reservation.start, Ticket.subtitle)
query = query.filter(Reservation.status == 'approved')
if exclude_pending:
query = query.filter(Reservation.data['accepted'] == True)
return query
[docs]
def reservation_title(self, reservation: Reservation) -> str:
title = self.title_template.format(
start=reservation.display_start(),
end=reservation.display_end(),
quota=reservation.quota
)
if title.endswith('00:00'):
return title[:-5] + '24:00'
return title
[docs]
class DaypassResource(Resource, AccessExtension, SearchableContent,
ContactExtension, PersonLinkExtension,
CoordinatesExtension, SharedMethods,
ResourceValidationExtension, GeneralFileLinkExtension,
InlinePhotoAlbumExtension):
[docs]
__mapper_args__ = {'polymorphic_identity': 'daypass'}
[docs]
fts_type_title = _('Resources')
# the selected view
# show or hide quota numbers in reports
# use to render the reservation title
[docs]
title_template = '{start:%d.%m.%Y} ({quota})'
[docs]
class RoomResource(Resource, AccessExtension, SearchableContent,
ContactExtension, PersonLinkExtension,
CoordinatesExtension, SharedMethods,
ResourceValidationExtension, GeneralFileLinkExtension,
InlinePhotoAlbumExtension):
[docs]
__mapper_args__ = {'polymorphic_identity': 'room'}
[docs]
fts_type_title = _('Resources')
# the selected view (depends on the resource's default)
# show or hide quota numbers in reports
# used to render the reservation title
[docs]
title_template = '{start:%d.%m.%Y} {start:%H:%M} - {end:%H:%M}'
[docs]
kaba_components: dict_property[list[tuple[str, str]]] = (
meta_property(default=list)
)
@property
[docs]
def has_future_reservations(self) -> bool:
future = self.scheduler.session.query(Reservation).filter(
Reservation.resource.in_([self.id, self.virtual_id])
).filter(Reservation.end >= sedate.utcnow())
return self.scheduler.session.query(future.exists()).scalar()
@property
[docs]
def has_future_reserved_slots(self) -> Query[ReservedSlot]:
future = self.scheduler.session.query(ReservedSlot).filter(
ReservedSlot.resource.in_([self.id, self.virtual_id])
).filter(ReservedSlot.end >= sedate.utcnow())
return self.scheduler.session.query(future.exists()).scalar()
@property
[docs]
def deletable(self) -> bool:
if self.has_future_reserved_slots or self.has_future_reservations:
return False
return True
[docs]
class ItemResource(Resource, AccessExtension, SearchableContent,
ContactExtension, PersonLinkExtension,
CoordinatesExtension, SharedMethods,
ResourceValidationExtension, GeneralFileLinkExtension,
InlinePhotoAlbumExtension):
[docs]
__mapper_args__ = {'polymorphic_identity': 'daily-item'}
[docs]
fts_type_title = _('Resources')
[docs]
title_template = '{start:%d.%m.%Y} {start:%H:%M} - {end:%H:%M} ({quota})'
[docs]
kaba_components: dict_property[list[tuple[str, str]]] = (
meta_property(default=list)
)