Town6: API For News and Events

TYPE: Feature
LINK: OGC-1950

src/onegov/event/collections/occurrences.py

@@ -598,6 +598,12 @@ def by_name(self, name: str) -> Occurrence | None:
         query = self.session.query(Occurrence).filter(Occurrence.name == name)
         return query.first()
 
+    def by_id(self, id: str) -> Occurrence | None:
+        """ Returns an occurrence by its id. """
+
+        query = self.session.query(Occurrence).filter(Occurrence.id == id)
+        return query.first()
+
     def as_ical(self, request: CoreRequest) -> bytes:
         """ Returns the the events of the given occurrences as iCalendar
         string.

src/onegov/org/models/page.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from datetime import datetime
+from onegov.core.collection import Pagination
 from onegov.core.orm.mixins import (
     content_property, dict_markup_property, dict_property, meta_property)
 from onegov.form import Form, move_fields
@@ -22,6 +23,7 @@
 from onegov.org.models.traitinfo import TraitInfo
 from onegov.org.observer import observes
 from onegov.page import Page
+from onegov.page.collection import AdjacencyListCollection, PageCollection
 from onegov.search import SearchableContent
 from sedate import replace_timezone
 from sqlalchemy import desc, func, or_, and_
@@ -33,6 +35,7 @@
 if TYPE_CHECKING:
     from onegov.org.request import OrgRequest, PageMeta
     from sqlalchemy.orm import Query, Session
+    from typing import Self
 
 
 class Topic(Page, TraitInfo, SearchableContent, AccessExtension,
@@ -348,6 +351,50 @@ def all_tags(self) -> list[str]:
         return sorted(all_hashtags)
 
 
+class NewsCollection(Pagination[News], AdjacencyListCollection[News]):
+    """
+    Use it like this:
+
+        from onegov.page import NewsCollection
+        news = NewsCollection(session)
+    """
+
+    __listclass__ = News
+
+    def __init__(
+        self,
+        session: Session,
+        page: int = 0,
+    ):
+        self.session = session
+        self.page = page
+
+    def subset(self) -> Query[News]:
+        parent = PageCollection(self.session).by_path(
+            '/news/', ensure_type='news')
+        news = self.session.query(News)
+        if parent:
+            news = news.filter(Page.parent_id == parent.id)
+        news = news.filter(
+            News.publication_started == True,
+            News.publication_ended == False
+        )
+        news = news.order_by(desc(News.published_or_created))
+        news = news.options(undefer('created'))
+        news = news.options(undefer('content'))
+        return news
+
+    @property
+    def page_index(self) -> int:
+        return self.page
+
+    def page_by_index(self, index: int) -> Self:
+        return self.__class__(
+            self.session,
+            page=index
+        )
+
+
 class AtoZPages(AtoZ[Topic]):
 
     def get_title(self, item: Topic) -> str:

src/onegov/town6/api.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from onegov.event.collections import OccurrenceCollection
+from onegov.api import ApiEndpoint
+from onegov.gis import Coordinates
+
+
+from typing import Any
+from typing import TYPE_CHECKING
+
+from onegov.org.models.page import NewsCollection
+
+if TYPE_CHECKING:
+    from onegov.org.models.page import News
+    from onegov.town6.app import TownApp
+    from onegov.event.models import Occurrence
+    from onegov.core.orm.mixins import ContentMixin
+    from onegov.core.orm.mixins import TimestampMixin
+    from typing import TypeVar
+
+    T = TypeVar('T')
+
+
+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}
+
+
+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()
+
+
+class EventApiEndpoint(
+    ApiEndpoint['Occurrence'],
+):
+    app: TownApp
+    endpoint = 'events'
+
+    @property
+    def collection(self) -> Any:
+        result = OccurrenceCollection(
+            self.session,
+            page=self.page or 0
+        )
+
+        result.batch_size = self.batch_size
+        return result
+
+    def item_data(self, item: Occurrence) -> dict[str, Any]:
+        return {
+            'title': item.title,
+            'description': item.event.description,
+            'organizer': item.event.organizer,
+            'organizer_email': item.event.organizer_email,
+            'organizer_phone': item.event.organizer_phone,
+            'external_event_url': item.event.external_event_url,
+            'event_registration_url': item.event.event_registration_url,
+            'price': item.event.price,
+            'tags': item.event.tags,
+            'start': item.start.isoformat(),
+            'end': item.end.isoformat(),
+            'location': item.location,
+            'coordinates': get_geo_location(item),
+            'created': item.created.isoformat(),
+            'modified': get_modified_iso_format(item),
+        }
+
+    def item_links(self, item: Occurrence) -> dict[str, Any]:
+        return {
+            'image': item.event.image,
+            'pfd': item.event.pdf
+        }
+
+
+class NewsApiEndpoint(
+    ApiEndpoint['News'],
+):
+    app: TownApp
+    endpoint = 'news'
+    filters = set()
+
+    @property
+    def collection(self) -> Any:
+        result = NewsCollection(
+            self.session,
+            page=self.page or 0
+        )
+        result.batch_size = 25
+        return result
+
+    def item_data(self, item: News) -> dict[str, Any]:
+        if item.publication_start:
+            publication_start = item.publication_start.isoformat()
+        else:
+            publication_start = None
+
+        if item.publication_end:
+            publication_end = item.publication_end.isoformat()
+        else:
+            publication_end = None
+
+        return {
+            'title': item.title,
+            'lead': item.lead,
+            'text': item.text,
+            'publication_start': publication_start,
+            'publication_end': publication_end,
+            'created': item.created.isoformat(),
+            'modified': get_modified_iso_format(item),
+        }
+
+    def item_links(self, item: News) -> dict[str, Any]:
+        return {
+            'image': item.page_image,
+        }

src/onegov/town6/app.py

@@ -5,12 +5,14 @@
 import pytz
 from sedate import replace_timezone
 
+from onegov.api import ApiApp
 from onegov.core import utils
 from onegov.core.i18n import default_locale_negotiator
 from onegov.core.utils import module_path
 from onegov.foundation6.integration import FoundationApp
 from onegov.org.app import OrgApp
 from onegov.org.app import get_i18n_localedirs as get_org_i18n_localedirs
+from onegov.town6.api import EventApiEndpoint, NewsApiEndpoint
 from onegov.town6.custom import get_global_tools
 from onegov.town6.initial_content import create_new_organisation
 from onegov.town6.theme import TownTheme
@@ -19,12 +21,13 @@
 from typing import Any, TYPE_CHECKING
 if TYPE_CHECKING:
     from collections.abc import Callable, Iterator, Sequence
+    from onegov.api import ApiEndpoint
     from onegov.core.types import RenderData
     from onegov.org.models import Organisation
     from onegov.town6.request import TownRequest
 
 
-class TownApp(OrgApp, FoundationApp):
+class TownApp(OrgApp, FoundationApp, ApiApp):
 
     def configure_organisation(
         self,
@@ -190,6 +193,14 @@ def get_public_ticket_messages() -> tuple[str, ...]:
     )
 
 
+@TownApp.setting(section='api', name='endpoints')
+def get_api_endpoints() -> list[type[ApiEndpoint[Any]]]:
+    return [
+        EventApiEndpoint,
+        NewsApiEndpoint
+    ]
+
+
 @TownApp.webasset_path()
 def get_js_path() -> str:
     return 'assets/js'