mirror of
https://github.com/lucaspalomodevelop/eventcally.git
synced 2026-03-13 00:07:22 +00:00
322 lines
9.5 KiB
Python
322 lines
9.5 KiB
Python
from typing import Type
|
|
|
|
from dateutil.relativedelta import relativedelta
|
|
from flask import request
|
|
from sqlalchemy import and_
|
|
|
|
from project.dateutils import (
|
|
date_set_end_of_day,
|
|
form_input_from_date,
|
|
form_input_to_date,
|
|
get_today,
|
|
)
|
|
from project.models.trackable_mixin import TrackableMixin
|
|
|
|
|
|
class BaseSearchParams(object):
|
|
def __init__(self):
|
|
self.sort = None
|
|
|
|
def load_from_request(self, **kwargs):
|
|
self.sort = kwargs.get("sort", self.sort)
|
|
|
|
|
|
class TrackableSearchParams(BaseSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.created_at_from = None
|
|
self.created_at_to = None
|
|
|
|
def load_from_request(self, **kwargs):
|
|
super().load_from_request(**kwargs)
|
|
|
|
self.created_at_from = kwargs.get("created_at_from", self.created_at_from)
|
|
self.created_at_to = kwargs.get("created_at_to", self.created_at_to)
|
|
|
|
def get_trackable_query(self, query, klass: Type[TrackableMixin]):
|
|
filter = self.fill_trackable_filter(1 == 1, klass)
|
|
return query.filter(filter)
|
|
|
|
def fill_trackable_filter(self, filter, klass: Type[TrackableMixin]):
|
|
if self.created_at_from:
|
|
filter = and_(filter, klass.created_at >= self.created_at_from)
|
|
|
|
if self.created_at_to:
|
|
filter = and_(filter, klass.created_at < self.created_at_to)
|
|
|
|
return filter
|
|
|
|
def get_trackable_order_by(self, query, klass: Type[TrackableMixin]):
|
|
if self.sort == "-created_at":
|
|
query = query.order_by(klass.created_at.desc())
|
|
elif self.sort == "-updated_at":
|
|
query = query.order_by(klass.updated_at.desc().nulls_last())
|
|
elif self.sort == "-last_modified_at":
|
|
query = query.order_by(klass.last_modified_at.desc())
|
|
|
|
return query
|
|
|
|
|
|
class EventReferenceSearchParams(TrackableSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.admin_unit_id = None
|
|
|
|
|
|
class EventReferenceRequestSearchParams(TrackableSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.admin_unit_id = None
|
|
self.review_status = None
|
|
|
|
|
|
class AdminUnitVerificationRequestSearchParams(TrackableSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.source_admin_unit_id = None
|
|
self.target_admin_unit_id = None
|
|
self.review_status = None
|
|
|
|
|
|
class AdminUnitSearchParams(TrackableSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.keyword = None
|
|
self.include_unverified = False
|
|
self.only_verifier = False
|
|
self.reference_request_for_admin_unit_id = None
|
|
self.incoming_verification_requests_postal_code = None
|
|
|
|
def load_from_request(self, **kwargs):
|
|
super().load_from_request(**kwargs)
|
|
|
|
self.keyword = kwargs.get("keyword", self.keyword)
|
|
|
|
|
|
class OrganizerSearchParams(TrackableSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.admin_unit_id = None
|
|
self.name = None
|
|
|
|
def load_from_request(self, **kwargs):
|
|
super().load_from_request(**kwargs)
|
|
|
|
self.name = kwargs.get("name", self.name)
|
|
|
|
|
|
class EventPlaceSearchParams(TrackableSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.admin_unit_id = None
|
|
self.name = None
|
|
|
|
def load_from_request(self, **kwargs):
|
|
super().load_from_request(**kwargs)
|
|
|
|
self.name = kwargs.get("name", self.name)
|
|
|
|
|
|
class EventSearchParams(TrackableSearchParams):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._date_from = None
|
|
self._date_to = None
|
|
self._date_from_str = None
|
|
self._date_to_str = None
|
|
self._coordinate = None
|
|
self.admin_unit_id = None
|
|
self.include_admin_unit_references = None
|
|
self.admin_unit_references_only = None
|
|
self.can_read_private_events = None
|
|
self.can_read_planned_events = None
|
|
self.keyword = None
|
|
self.latitude = None
|
|
self.longitude = None
|
|
self.distance = None
|
|
self.category_id = None
|
|
self.organizer_id = None
|
|
self.event_place_id = None
|
|
self.event_list_id = None
|
|
self.weekday = None
|
|
self.status = None
|
|
self.public_status = None
|
|
self.favored_by_user_id = None
|
|
self.postal_code = None
|
|
self.not_referenced_by_organization_id = None
|
|
self.exclude_recurring = False
|
|
self.expected_participants_min = None
|
|
self.sort = "start"
|
|
|
|
@property
|
|
def date_from(self):
|
|
return self._date_from
|
|
|
|
@date_from.setter
|
|
def date_from(self, value):
|
|
self._date_from = value
|
|
self._date_from_str = form_input_from_date(value)
|
|
|
|
@property
|
|
def date_to(self):
|
|
return self._date_to
|
|
|
|
@date_to.setter
|
|
def date_to(self, value):
|
|
self._date_to = value
|
|
self._date_to_str = form_input_from_date(value)
|
|
|
|
@property
|
|
def date_from_str(self):
|
|
return self._date_from_str
|
|
|
|
@date_from_str.setter
|
|
def date_from_str(self, value):
|
|
self._date_from_str = value
|
|
self._date_from = form_input_to_date(value)
|
|
|
|
@property
|
|
def date_to_str(self):
|
|
return self._date_to_str
|
|
|
|
@date_to_str.setter
|
|
def date_to_str(self, value):
|
|
self._date_to_str = value
|
|
self._date_to = form_input_to_date(value, 23, 59, 59)
|
|
|
|
@property
|
|
def coordinate(self):
|
|
return self._coordinate
|
|
|
|
@coordinate.setter
|
|
def coordinate(self, value):
|
|
self._coordinate = value
|
|
if value is not None and len(value) > 0:
|
|
(self.latitude, self.longitude) = value.split(",")
|
|
else:
|
|
self.latitude = None
|
|
self.longitude = None
|
|
|
|
def set_default_date_range(self):
|
|
today = get_today()
|
|
self.date_from = today
|
|
self.date_to = None
|
|
|
|
def set_planning_date_range(self):
|
|
today = get_today()
|
|
self.date_from = today
|
|
self.date_to = date_set_end_of_day(today + relativedelta(months=3))
|
|
|
|
def load_list_param(self, param: str):
|
|
item_ids = request.args.getlist(param)
|
|
|
|
if len(item_ids) == 1 and "," in item_ids[0]:
|
|
item_ids = [i.strip() for i in item_ids[0].split(",")]
|
|
|
|
if "0" in item_ids:
|
|
item_ids.remove("0")
|
|
|
|
if len(item_ids) > 0:
|
|
return item_ids
|
|
|
|
return None
|
|
|
|
def load_bool_param(self, param: str):
|
|
return request.args[param].lower() in ("true", "t", "yes", "y", "on", "1")
|
|
|
|
def load_status_list_param(self):
|
|
stati = self.load_list_param("status")
|
|
|
|
if stati is None: # pragma: no cover
|
|
return None
|
|
|
|
from project.models import EventStatus
|
|
|
|
result = list()
|
|
|
|
for status in stati:
|
|
if status in EventStatus.__members__:
|
|
result.append(EventStatus.__members__[status])
|
|
|
|
return result
|
|
|
|
def load_public_status_list_param(self):
|
|
public_stati = self.load_list_param("public_status")
|
|
|
|
if public_stati is None: # pragma: no cover
|
|
return None
|
|
|
|
from project.models import PublicStatus
|
|
|
|
result = list()
|
|
|
|
for public_status in public_stati:
|
|
if public_status in PublicStatus.__members__:
|
|
result.append(PublicStatus.__members__[public_status])
|
|
|
|
return result
|
|
|
|
def load_from_request(self, **kwargs):
|
|
super().load_from_request(**kwargs)
|
|
|
|
if "date_from" in request.args:
|
|
self.date_from_str = request.args["date_from"]
|
|
|
|
if "date_to" in request.args:
|
|
self.date_to_str = request.args["date_to"]
|
|
|
|
if "keyword" in request.args:
|
|
self.keyword = request.args["keyword"]
|
|
|
|
if "coordinate" in request.args:
|
|
self.coordinate = request.args["coordinate"]
|
|
|
|
if "distance" in request.args:
|
|
self.distance = request.args["distance"]
|
|
|
|
if "category_id" in request.args:
|
|
self.category_id = self.load_list_param("category_id")
|
|
|
|
if "weekday" in request.args:
|
|
self.weekday = request.args.getlist("weekday")
|
|
|
|
if "organizer_id" in request.args:
|
|
self.organizer_id = request.args["organizer_id"]
|
|
|
|
if "event_place_id" in request.args:
|
|
self.event_place_id = request.args["event_place_id"]
|
|
|
|
if "expected_participants_min" in request.args:
|
|
self.expected_participants_min = request.args["expected_participants_min"]
|
|
|
|
if "event_list_id" in request.args:
|
|
self.event_list_id = self.load_list_param("event_list_id")
|
|
|
|
if "postal_code" in request.args:
|
|
self.postal_code = self.load_list_param("postal_code")
|
|
|
|
if "organization_id" in request.args:
|
|
self.admin_unit_id = request.args["organization_id"]
|
|
|
|
if "admin_unit_id" in request.args:
|
|
self.admin_unit_id = request.args["admin_unit_id"]
|
|
|
|
if "status" in request.args:
|
|
self.status = self.load_status_list_param()
|
|
|
|
if "public_status" in request.args:
|
|
self.public_status = self.load_public_status_list_param()
|
|
|
|
if "exclude_recurring" in request.args:
|
|
self.exclude_recurring = self.load_bool_param("exclude_recurring")
|
|
|
|
if "include_organization_references" in request.args:
|
|
self.include_admin_unit_references = self.load_bool_param(
|
|
"include_organization_references"
|
|
)
|
|
|
|
if "organization_references_only" in request.args:
|
|
self.admin_unit_references_only = self.load_bool_param(
|
|
"organization_references_only"
|
|
)
|