from dateutil.parser import isoparse
from functools import cached_property
from onegov.agency.collections import ExtendedPersonCollection
from onegov.agency.collections import PaginatedAgencyCollection
from onegov.agency.collections import PaginatedMembershipCollection
from onegov.agency.forms import PersonMutationForm
from onegov.api import ApiEndpoint, ApiInvalidParamException
from onegov.gis import Coordinates
from typing import Any
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from onegov.agency.app import AgencyApp
from onegov.agency.models import ExtendedAgency
from onegov.agency.models import ExtendedAgencyMembership
from onegov.agency.models import ExtendedPerson
from onegov.core.orm.mixins import ContentMixin
from onegov.core.orm.mixins import TimestampMixin
from typing import TypeVar
[docs]
UPDATE_FILTER_PARAMS = frozenset((
'updated_gt',
'updated_lt',
'updated_eq',
'updated_ge',
'updated_le'
))
[docs]
def filter_for_updated(
filter_operation: str,
filter_value: str | None,
result: 'T'
) -> 'T':
"""
Applies filters for several 'updated' comparisons.
Refer to UPDATE_FILTER_PARAMS for all filter keywords.
:param filter_operation: the updated filter operation to be applied. For
allowed filters refer to UPDATE_FILTER_PARAMS
:param filter_value: the updated filter value to filter for
:param result: the results to apply the filters on
:return: filter result
"""
assert hasattr(result, 'for_filter')
if filter_value is None:
return result.for_filter(**{filter_operation: None})
try:
# only parse including hours and minutes
isoparse(filter_value[:16])
except Exception as ex:
raise ApiInvalidParamException(f'Invalid iso timestamp for parameter'
f"'{filter_operation}': {ex}") from ex
return result.for_filter(**{filter_operation: filter_value[:16]})
[docs]
class ApisMixin:
@cached_property
[docs]
def agency_api(self) -> 'AgencyApiEndpoint':
return AgencyApiEndpoint(self.app)
@cached_property
[docs]
def person_api(self) -> 'PersonApiEndpoint':
return PersonApiEndpoint(self.app)
@cached_property
[docs]
def membership_api(self) -> 'MembershipApiEndpoint':
return MembershipApiEndpoint(self.app)
[docs]
def get_geo_location(item: 'ContentMixin') -> dict[str, Any]:
geo = item.content.get('coordinates', Coordinates()) or Coordinates()
return {'lon': geo.lon, 'lat': geo.lat, 'zoom': geo.zoom}
[docs]
class PersonApiEndpoint(ApiEndpoint['ExtendedPerson'], ApisMixin):
[docs]
filters = {'first_name', 'last_name'} | UPDATE_FILTER_PARAMS
@property
[docs]
def collection(self) -> ExtendedPersonCollection:
result = ExtendedPersonCollection(
self.session,
page=self.page or 0
)
for key, value in self.extra_parameters.items():
self.assert_valid_filter(key)
# apply different filters
if key == 'first_name':
result = result.for_filter(first_name=value)
elif key == 'last_name':
result = result.for_filter(last_name=value)
elif key in UPDATE_FILTER_PARAMS:
result = filter_for_updated(filter_operation=key,
filter_value=value,
result=result)
result.exclude_hidden = True
result.batch_size = self.batch_size
return result
[docs]
def item_data(self, item: 'ExtendedPerson') -> dict[str, Any]:
data = {
attribute: getattr(item, attribute, None)
for attribute in (
'academic_title',
'born',
'email',
'first_name',
'function',
'last_name',
'location_address',
'location_code_city',
'notes',
'parliamentary_group',
'phone',
'phone_direct',
'political_party',
'postal_address',
'postal_code_city',
'profession',
'salutation',
'title',
'website',
)
if attribute not in self.app.org.hidden_people_fields
}
data['modified'] = get_modified_iso_format(item)
return data
[docs]
def item_links(self, item: 'ExtendedPerson') -> dict[str, Any]:
result = {
attribute: getattr(item, attribute, None)
for attribute in (
'picture_url',
'website',
)
if attribute not in self.app.org.hidden_people_fields
}
result['memberships'] = self.membership_api.for_filter(
person=item.id.hex
)
return result
[docs]
def apply_changes(
self,
item: 'ExtendedPerson',
form: PersonMutationForm
) -> None:
# FIXME: circular import
from onegov.agency.views.people import do_report_person_change
do_report_person_change(item, form.meta.request, form)
[docs]
class AgencyApiEndpoint(ApiEndpoint['ExtendedAgency'], ApisMixin):
[docs]
filters = {'parent', 'title'} | UPDATE_FILTER_PARAMS
@property
[docs]
def collection(self) -> PaginatedAgencyCollection:
result = PaginatedAgencyCollection(
self.session,
page=self.page or 0,
parent=self.get_filter('parent', None, False),
joinedload=['organigram'],
undefer=['content']
)
for key, value in self.extra_parameters.items():
self.assert_valid_filter(key)
# apply different filters
if key == 'title':
result = result.for_filter(title=value)
elif key in UPDATE_FILTER_PARAMS:
result = filter_for_updated(filter_operation=key,
filter_value=value,
result=result)
result.batch_size = self.batch_size
return result
[docs]
def item_data(self, item: 'ExtendedAgency') -> dict[str, Any]:
return {
'title': item.title,
'portrait': item.portrait,
'location_address': item.location_address,
'location_code_city': item.location_code_city,
'modified': get_modified_iso_format(item),
'postal_address': item.postal_address,
'postal_code_city': item.postal_code_city,
'website': item.website,
'email': item.email,
'phone': item.phone,
'phone_direct': item.phone_direct,
'opening_hours': item.opening_hours,
'geo_location': get_geo_location(item),
}
[docs]
def item_links(self, item: 'ExtendedAgency') -> dict[str, Any]:
return {
'organigram': item.organigram,
'parent': self.for_item(item.parent),
'children': self.for_filter(parent=str(item.id)),
'memberships': self.membership_api.for_filter(
agency=str(item.id)
)
}
[docs]
class MembershipApiEndpoint(
ApiEndpoint['ExtendedAgencyMembership'],
ApisMixin
):
[docs]
endpoint = 'memberships'
[docs]
filters = {'agency', 'person'} | UPDATE_FILTER_PARAMS
@property
[docs]
def collection(self) -> PaginatedMembershipCollection:
result = PaginatedMembershipCollection(
self.session,
page=self.page or 0,
agency=self.get_filter('agency'),
person=self.get_filter('person'),
)
for key, value in self.extra_parameters.items():
self.assert_valid_filter(key)
# apply different filters
if key in UPDATE_FILTER_PARAMS:
result = filter_for_updated(filter_operation=key,
filter_value=value,
result=result)
result.batch_size = self.batch_size
return result
[docs]
def item_data(self, item: 'ExtendedAgencyMembership') -> dict[str, Any]:
return {
'title': item.title,
'modified': get_modified_iso_format(item),
}
[docs]
def item_links(self, item: 'ExtendedAgencyMembership') -> dict[str, Any]:
return {
'agency': self.agency_api.for_item(item.agency),
'person': self.person_api.for_item(item.person)
}