from __future__ import annotations
from datetime import datetime
from sedate import utcnow
from sqlalchemy import desc
from onegov.core.collection import Pagination, GenericCollection
from onegov.fsi.collections.course import CourseCollection
from onegov.fsi.collections.notification_template import (
CourseNotificationTemplateCollection)
from onegov.fsi.models.course import Course
from onegov.fsi.models.course_event import CourseEvent
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from sqlalchemy.orm import Query, Session
from typing import Self
from uuid import UUID
[docs]
class CourseEventCollection(
GenericCollection[CourseEvent],
Pagination[CourseEvent]
):
def __init__(
self,
session: Session,
page: int = 0,
from_date: datetime | None = None,
upcoming_only: bool = False,
past_only: bool = False,
limit: int | None = None,
show_hidden: bool = False,
show_locked: bool = True,
course_id: UUID | None = None,
sort_desc: bool = False
) -> None:
super().__init__(session)
# filter newer than from date
[docs]
self.from_date = from_date # ignores upcoming_only
[docs]
self.upcoming_only = upcoming_only # active if from_date not set
[docs]
self.past_only = past_only
[docs]
self.show_hidden = show_hidden
[docs]
self.show_locked = show_locked
[docs]
self.course_id = course_id
[docs]
self.sort_desc = sort_desc
if from_date:
assert isinstance(from_date, datetime)
[docs]
def __eq__(self, other: object) -> bool:
return (
isinstance(other, self.__class__)
and self.page == other.page
and self.from_date == other.from_date
and self.upcoming_only == other.upcoming_only
and self.past_only == other.past_only
and self.limit == other.limit
and self.show_hidden == other.show_hidden
and self.course_id == other.course_id
and self.sort_desc == other.sort_desc
)
@property
[docs]
def model_class(self) -> type[CourseEvent]:
return CourseEvent
@property
[docs]
def course(self) -> Course | None:
if not self.course_id:
return None
return CourseCollection(self.session).by_id(self.course_id)
[docs]
def query(self) -> Query[CourseEvent]:
query = super().query()
if not self.show_hidden:
query = query.filter(CourseEvent.hidden_from_public == False)
query = query.join(Course)
query = query.filter(Course.hidden_from_public == False)
if not self.show_locked:
query = query.filter(CourseEvent.locked_for_subscriptions == False)
if self.from_date:
query = query.filter(CourseEvent.start > self.from_date)
elif self.past_only:
query = query.filter(CourseEvent.start <= utcnow())
elif self.upcoming_only:
query = query.filter(CourseEvent.start >= utcnow())
if self.course_id:
query = query.filter(CourseEvent.course_id == self.course_id)
ordering = CourseEvent.start
query = query.order_by(desc(ordering) if self.sort_desc else ordering)
if self.limit:
query = query.limit(self.limit)
return query
[docs]
def subset(self) -> Query[CourseEvent]:
return self.query()
@property
[docs]
def page_index(self) -> int:
return self.page
[docs]
def page_by_index(self, index: int) -> Self:
return self.__class__(
self.session,
page=index,
from_date=self.from_date,
upcoming_only=self.upcoming_only,
past_only=self.past_only,
limit=self.limit,
show_hidden=self.show_hidden,
show_locked=self.show_locked,
course_id=self.course_id,
sort_desc=self.sort_desc
)
@classmethod
[docs]
def latest(cls, session: Session, limit: int = 5) -> Self:
return cls(session, upcoming_only=True, limit=limit)
[docs]
def next_event(self) -> Query[CourseEvent]:
return self.query().filter(
self.model_class.start > utcnow()).order_by(None).order_by(
self.model_class.start)
[docs]
def get_past_reminder_date(self) -> Query[CourseEvent]:
return super().query().filter(
self.model_class.scheduled_reminder > utcnow())
[docs]
def add(self, **kwargs: Any) -> CourseEvent:
# store the course instead of the course_id, for Elasticsearch to
# properly work (which needs access to the course)
course = self.session.query(Course).filter_by(
id=kwargs['course_id']).one()
course_event = super().add(course=course, **kwargs)
tc = CourseNotificationTemplateCollection(
self.session, course_event_id=course_event.id)
tc.auto_add_templates_if_not_existing()
return course_event
[docs]
class PastCourseEventCollection(CourseEventCollection):
"""This is used for past events to do the audit """
def __init__(
self,
session: Session,
page: int = 0,
show_hidden: bool = False,
show_locked: bool = True,
course_id: UUID | None = None
) -> None:
super().__init__(
session,
page=page,
past_only=True,
show_hidden=show_hidden,
show_locked=show_locked,
course_id=course_id,
sort_desc=True
)
[docs]
def page_by_index(self, index: int) -> Self:
return self.__class__(
self.session,
page=index,
show_hidden=self.show_hidden,
show_locked=self.show_locked,
course_id=self.course_id,
)
[docs]
def query(self) -> Query[CourseEvent]:
return super().query().filter(self.model_class.status == 'confirmed')