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.core.orm.types import UUID
from onegov.form.models import FormSubmission
from onegov.org import _
from onegov.org.models.extensions import (
ContactExtension, GeneralFileLinkExtension, 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.orm import undefer
from sqlalchemy.sql.expression import cast
from uuid import uuid4, uuid5
from typing import Any, ClassVar, TYPE_CHECKING
if TYPE_CHECKING:
import uuid
from libres.context.core import Context
from libres.db.scheduler import Scheduler
from onegov.org.request import OrgRequest
from sqlalchemy import Column
from sqlalchemy.orm import Query
[docs]
class FindYourSpotCollection(ResourceCollection):
def __init__(self, libres_context: 'Context', group: str | None) -> None:
super().__init__(libres_context)
@property
[docs]
def title(self) -> str:
return _('Find Your Spot')
@property
[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 ''))
return query
[docs]
class SharedMethods:
if TYPE_CHECKING:
[docs]
title_template: ClassVar[str]
id: Column[uuid.UUID]
libres_context: Context
date: date | None
view: str | None
timezone: Column[str]
@property
def scheduler(self) -> Scheduler: ...
def get_scheduler(self, context: Context) -> Scheduler: ...
[docs]
lead: dict_property[str | None] = meta_property()
[docs]
text = dict_markup_property('content')
[docs]
occupancy_is_visible_to_members: dict_property[bool | None]
occupancy_is_visible_to_members = 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 future_managed_reservations(self) -> 'Query[Reservation]':
return self.scheduler.managed_reservations().filter( # type:ignore
Reservation.end >= sedate.utcnow())
@property
[docs]
def future_managed_reserved_slots(self) -> 'Query[ReservedSlot]':
return self.scheduler.managed_reserved_slots().filter(
ReservedSlot.end >= sedate.utcnow())
@property
[docs]
def calendar_date_range(self) -> tuple[datetime, datetime]:
""" Returns 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 == 'month':
return sedate.align_range_to_month(date, date, self.timezone)
elif self.view == 'agendaWeek':
return sedate.align_range_to_week(date, date, self.timezone)
elif self.view == 'agendaDay':
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:
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.all()}
query = session.query(FormSubmission)
query = query.filter(FormSubmission.name == None)
query = query.filter(FormSubmission.id.in_(tokens))
query.delete('fetch')
queries.remove_expired_reservation_sessions(expiration_date)
[docs]
def bound_reservations(
self,
request: 'OrgRequest',
status: str = 'pending'
) -> '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)
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))
return res # type:ignore[return-value]
[docs]
def bound_session_id(self, request: 'OrgRequest') -> 'uuid.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
) -> 'Query[Reservation]':
""" Returns a query which joins this resource's reservations between
start and end with the tickets table.
"""
query = self.scheduler.managed_reservations()
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, UUID))
query = query.order_by(Reservation.start)
query = query.order_by(Ticket.subtitle)
query = query.filter(Reservation.status == 'approved')
if exclude_pending:
query = query.filter(Reservation.data != None)
return query # type:ignore[return-value]
[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):
[docs]
__mapper_args__ = {'polymorphic_identity': 'daypass'}
[docs]
es_type_name = 'daypasses'
# 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):
[docs]
__mapper_args__ = {'polymorphic_identity': 'room'}
# 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}'
@property
[docs]
def deletable(self) -> bool:
if self.future_managed_reserved_slots.first():
return False
if self.future_managed_reservations.first():
return False
return True
[docs]
class ItemResource(Resource, AccessExtension, SearchableContent,
ContactExtension, PersonLinkExtension,
CoordinatesExtension, SharedMethods,
ResourceValidationExtension, GeneralFileLinkExtension):
[docs]
__mapper_args__ = {'polymorphic_identity': 'daily-item'}
[docs]
es_type_name = 'daily_items'
[docs]
title_template = '{start:%d.%m.%Y} {start:%H:%M} - {end:%H:%M} ({quota})'