Integrate pygeoapi

geonode_dominode/dominode_pygeoapi/apps.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class DominodePygeoapiConfig(AppConfig):
+    name = 'dominode_pygeoapi'

geonode_dominode/dominode_pygeoapi/forms.py

@@ -0,0 +1,101 @@
+import datetime as dt
+import typing
+
+from django import forms
+from django.utils.translation import gettext_lazy as _
+
+
+class GeojsonField(forms.CharField):
+
+    def validate(self, value: str) -> typing.Dict:
+        # TODO: convert input to GeoJSON
+        return value
+
+
+class BboxField(forms.MultiValueField):
+
+    def __init__(self, **kwargs):
+        fields = (
+            forms.FloatField(help_text='lower left corner, coordinate axis 1'),
+            forms.FloatField(help_text='lower left corner, coordinate axis 2'),
+            forms.FloatField(
+                required=False,
+                help_text='minimum value, coordinate axis 3'
+            ),
+            forms.FloatField(help_text='upper right corner, coordinate axis 1'),
+            forms.FloatField(help_text='upper right corner, coordinate axis 2'),
+            forms.FloatField(
+                required=False,
+                help_text='maximum value, coordinate axis 3'
+            ),
+        )
+        super().__init__(fields=fields, require_all_fields=False, **kwargs)
+
+    def compress(self, data_list):
+        # TODO: return a geojson polygon
+        return data_list
+
+
+class StringListField(forms.CharField):
+
+    def __init__(self, separator: typing.Optional[str] = ',', **kwargs):
+        super().__init__(**kwargs)
+        self.separator = separator
+
+    def validate(self, value: str) -> typing.List[str]:
+        return [i.strip() for i in value.split(self.separator)]
+
+
+class StacDatetimeField(forms.Field):
+
+    def validate(
+            self,
+            value: str
+    ) -> typing.List[typing.Union[str, dt.datetime]]:
+        # can either be a single datetime, a closed interval or an open interval
+        open_side = '..'
+        parts = value.split('/')
+        result = []
+        if len(parts) == 1:
+            # a single value
+            # FIXME: guard this with a try block
+            result.append(dt.datetime.fromisoformat(parts[0]))
+        elif len(parts) == 2:
+            start, end = parts
+            if start == open_side and end == open_side:
+                raise forms.ValidationError(
+                    _('Interval cannot be open on both sides'),
+                    code='invalid',
+                )
+            elif start == open_side:
+                result.append(start)
+            else:
+                result.append(dt.datetime.fromisoformat(start))
+            if end == open_side:
+                result.append(end)
+            else:
+                result.append(dt.datetime.fromisoformat(end))
+        else:
+            raise forms.ValidationError(
+                _('Invalid datetime format'),
+                code='invalid'
+            )
+        return result
+
+
+class StacSearchSimpleForm(forms.Form):
+    bbox = BboxField(required=False)
+    bbox_crs = forms.CharField(label='bbox-crs', required=False)
+    datetime_ = StacDatetimeField(label='datetime', required=False)
+    limit = forms.IntegerField(
+        min_value=1,
+        max_value=10000,
+        initial=10,
+        required=False
+    )
+    ids = StringListField(required=False)
+    collections = StringListField(required=False)
+
+
+class StacSearchCompleteForm(StacSearchSimpleForm):
+    intersects = GeojsonField(required=False)

geonode_dominode/dominode_pygeoapi/urls.py

@@ -0,0 +1,74 @@
+from django.urls import (
+    include,
+    path,
+)
+
+from . import views
+
+urlpatterns = [
+    path(
+        '',
+        views.pygeoapi_root,
+        name='pygeoapi-root'
+    ),
+    path(
+        'openapi/',
+        views.pygeoapi_openapi_endpoint,
+        name='pygeoapi-openapi'
+    ),
+    path(
+        'conformance/',
+        views.pygeoapi_conformance_endpoint,
+        name='pygeoapi-conformance'
+    ),
+    path(
+        'collections/',
+        views.pygeoapi_collections_endpoint,
+        name='pygeoapi-collection-list'
+    ),
+    path(
+        'collections/<str:name>/',
+        views.pygeoapi_collections_endpoint,
+        name='pygeoapi-collection-detail'
+    ),
+    path(
+        'collections/<str:name>/queryables/',
+        views.pygeoapi_collection_queryables,
+        name='pygeoapi-collection-queryable-list'
+    ),
+    path(
+        'collections/<str:collection_id>/items/',
+        views.get_pygeoapi_dataset_list,
+        name='pygeoapi-collection-item-list'
+    ),
+    path(
+        'collections/<str:collection_id>/items/<str:item_id>/',
+        views.get_pygeoapi_dataset_detail,
+        name='pygeoapi-collection-item-detail'
+    ),
+    path(
+        'stac/',
+        views.stac_catalog_root,
+        name='pygeoapi-stac-catalog-root'
+    ),
+    path(
+        'stac/<path:path>',
+        views.stac_catalog_path_endpoint,
+        name='pygeoapi-stac-catalog-path'
+    ),
+    path(
+        'stac/search/',
+        views.stac_catalog_path,
+        name='pygeoapi-stac-search'
+    ),
+    path(
+        'processes/',
+        views.get_pygeoapi_processes,
+        name='pygeoapi-process-list'
+    ),
+    path(
+        'processes/<str:name>',
+        views.get_pygeoapi_processes,
+        name='pygeoapi-process-detail'
+    ),
+]

geonode_dominode/dominode_pygeoapi/views.py

@@ -0,0 +1,145 @@
+import datetime as dt
+"""Integration of pygeoapi into DomiNode"""
+import typing
+
+from django.conf import settings
+from django.http import (
+    HttpRequest,
+    HttpResponse
+)
+from django.views import View
+from pygeoapi.api import API
+from pygeoapi.openapi import get_oas
+
+from .forms import (
+    StacSearchCompleteForm,
+    StacSearchSimpleForm,
+)
+
+# TODO: test these views
+# TODO: add authentication
+# TODO: add authorization
+
+def pygeoapi_root(request: HttpRequest) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(request, 'root')
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def pygeoapi_openapi_endpoint(request: HttpRequest) -> HttpResponse:
+    openapi_config = get_oas(settings.PYGEOAPI_CONFIG)
+    pygeoapi_response = _get_pygeoapi_response(
+        request, 'openapi', openapi_config)
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def pygeoapi_conformance_endpoint(request: HttpRequest) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(request, 'conformance')
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def pygeoapi_collections_endpoint(
+        request: HttpRequest,
+        name: typing.Optional[str] = None,
+) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(
+        request, 'describe_collections', name)
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def pygeoapi_collection_queryables(
+        request: HttpRequest,
+        name: typing.Optional[str] = None,
+) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(
+        request, 'get_collection_queryables', name)
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def get_pygeoapi_dataset_list(
+        request: HttpRequest,
+        collection_id: str
+) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(
+        request, 'get_collection_items', collection_id)
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def get_pygeoapi_dataset_detail(
+        request: HttpRequest,
+        collection_id: str,
+        item_id: str,
+) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(
+        request, 'get_collection_item', collection_id, item_id)
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def stac_catalog_root(request: HttpRequest) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(request, 'get_stac_root')
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def stac_catalog_path_endpoint(
+        request: HttpRequest,
+        path: str
+) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(request, 'get_stac_path', path)
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+def get_pygeoapi_processes(
+        request: HttpRequest,
+        name: typing.Optional[str] = None
+) -> HttpResponse:
+    pygeoapi_response = _get_pygeoapi_response(
+        request, 'describe_processes', name)
+    return _convert_pygeoapi_response_to_django_response(*pygeoapi_response)
+
+
+class StacSearchView(View):
+
+    def get(self, request: HttpRequest) -> HttpResponse:
+        form_ = StacSearchSimpleForm(request.GET)
+        if form_.is_valid():
+            self._perform_stac_search()
+
+    def post(self, request: HttpRequest) -> HttpResponse:
+        form_ = StacSearchCompleteForm(request.POST)
+        if form_.is_valid():
+            self._perform_stac_search()
+
+    def _perform_stac_search(
+            self,
+            bbox: typing.Optional[typing.Any] = None,
+            bbox_crs: typing.Optional[str] = None,
+            datetime_: typing.Optional[dt.datetime] = None,
+            limit: typing.Optional[int] = None,
+            ids: typing.Optional[typing.List[str]] = None,
+            collections: typing.Optional[typing.List[str]] = None,
+            intersects: typing.Optional[typing.Any] = None,
+    ):
+        pass
+
+
+def _get_pygeoapi_response(
+        request: HttpRequest,
+        api_method_name: str,
+        *args,
+        **kwargs
+) -> typing.Tuple[typing.Dict, int, str]:
+    """Use pygeoapi to process the input request"""
+    pygeoapi_api = API(settings.PYGEOAPI_CONFIG)
+    api_method = getattr(pygeoapi_api, api_method_name)
+    return api_method(request.headers, request.GET, *args, **kwargs)
+
+
+def _convert_pygeoapi_response_to_django_response(
+        pygeoapi_headers: typing.Mapping,
+        pygeoapi_status_code: int,
+        pygeoapi_content: str,
+) -> HttpResponse:
+    """Convert pygeoapi response to a django response"""
+    response = HttpResponse(pygeoapi_content, status=pygeoapi_status_code)
+    for key, value in pygeoapi_headers.items():
+        response[key] = value
+    return response