Source code for agency.api

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.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] T = TypeVar('T')
[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:
[docs] app: 'AgencyApp'
@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] def get_modified_iso_format(item: 'TimestampMixin') -> str: """ Returns the iso format of the modified or created field of item. :param item: db item e.g. agency, people, membership :return: str iso representation of item last modification """ return item.last_change.isoformat()
[docs] class PersonApiEndpoint(ApiEndpoint['ExtendedPerson'], ApisMixin):
[docs] app: 'AgencyApp'
[docs] endpoint = 'people'
[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] class AgencyApiEndpoint(ApiEndpoint['ExtendedAgency'], ApisMixin):
[docs] app: 'AgencyApp'
[docs] endpoint = 'agencies'
[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] class MembershipApiEndpoint( ApiEndpoint['ExtendedAgencyMembership'], ApisMixin ):
[docs] app: 'AgencyApp'
[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), }