import enum
from onegov.core.utils import normalize_for_url
from onegov.reservation.models import Resource
from uuid import uuid4, UUID
from typing import overload, Any, Literal, TypeVar, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Callable
from libres.context.core import Context
from libres.db.models import Allocation, Reservation
from sqlalchemy.orm import Query
from typing import TypeAlias
[docs]
_R = TypeVar('_R', bound=Resource)
[docs]
class _Marker(enum.Enum):
[docs]
any_type_t: 'TypeAlias' = Literal[_Marker.any_type]
[docs]
any_type: any_type_t = _Marker.any_type
[docs]
class ResourceCollection:
""" Manages a list of resources.
"""
def __init__(self, libres_context: 'Context'):
assert hasattr(libres_context, 'get_service'), """
The ResourceCollection expected the libres_contex, not the session.
"""
[docs]
self.libres_context = libres_context
[docs]
self.session = libres_context.get_service('session_provider').session()
[docs]
def query(self) -> 'Query[Resource]':
return self.session.query(Resource)
[docs]
def add(
self,
title: str,
timezone: str,
type: str | None = None,
name: str | None = None,
meta: dict[str, Any] | None = None,
content: dict[str, Any] | None = None,
definition: str | None = None,
group: str | None = None
) -> Resource:
if type is None:
resource = Resource()
else:
# look up the right class depending on the type (we need to do
# this a bit akwardly here, because Resource does not use the
# ModelBase as declarative base)
resource = Resource.get_polymorphic_class(type, Resource)()
resource.id = uuid4()
resource.name = name or normalize_for_url(title)
resource.title = title
resource.timezone = timezone
resource.meta = meta or {}
resource.content = content or {}
resource.definition = definition
resource.group = group
resource.renew_access_token()
self.session.add(resource)
self.session.flush()
return self.bind(resource)
@overload
[docs]
def bind(self, resource: _R) -> _R: ...
@overload
def bind(self, resource: None) -> None: ...
def bind(self, resource: _R | None) -> _R | None:
if resource:
resource.bind_to_libres_context(self.libres_context)
return resource
[docs]
def by_id(
self,
id: UUID,
ensure_type: str | any_type_t = any_type
) -> Resource | None:
query = self.query().filter(Resource.id == id)
if ensure_type is not any_type:
query = query.filter(Resource.type == ensure_type)
return self.bind(query.first())
[docs]
def by_name(
self,
name: str,
ensure_type: str | any_type_t = any_type
) -> Resource | None:
query = self.query().filter(Resource.name == name)
if ensure_type is not any_type:
query = query.filter(Resource.type == ensure_type)
return self.bind(query.first())
[docs]
def by_allocation(self, allocation: 'Allocation') -> Resource | None:
return self.by_id(allocation.resource)
[docs]
def by_reservation(self, reservation: 'Reservation') -> Resource | None:
return self.by_id(reservation.resource)
[docs]
def delete(
self,
resource: Resource,
including_reservations: bool = False,
handle_reservation: 'Callable[[Reservation], Any] | None' = None
) -> None:
scheduler = resource.get_scheduler(self.libres_context)
if not including_reservations:
assert not scheduler.managed_reserved_slots().first()
assert not scheduler.managed_reservations().first()
scheduler.managed_allocations().delete('fetch')
else:
if callable(handle_reservation):
for res in scheduler.managed_reservations():
# e.g. create a ticket snapshot
handle_reservation(res)
scheduler.extinguish_managed_records()
if resource.files:
# unlink any linked files
resource.files = []
self.session.flush()
self.session.delete(resource)
self.session.flush()