add Elasticsearch process manager

.github/workflows/main.yml

@@ -137,6 +137,7 @@ jobs:
         pybabel compile -d locale -l fr
         python3 tests/load_oracle_data.py
         python3 tests/load_es_data.py tests/data/ne_110m_populated_places_simple.geojson geonameid
+        python3 tests/load_es_manager_data.py tests/data/test-manager.json
         python3 tests/load_opensearch_data.py tests/data/ne_110m_populated_places_simple.geojson geonameid
         python3 tests/load_mongo_data.py tests/data/ne_110m_populated_places_simple.geojson
         gunzip < tests/data/hotosm_bdi_waterways.sql.gz | psql postgresql://postgres:${{ secrets.DatabasePassword || 'postgres' }}@localhost:5432/test

docs/source/publishing/ogcapi-features.rst

@@ -104,7 +104,7 @@ To publish an Elasticsearch index, the following are required in your index:
 
 .. note::
 
-   For Elasticseach indexes that are password protect, a RFC1738 URL can be used as follows:
+   For Elasticsearch indexes that are password protected, a RFC1738 URL can be used as follows:
 
    ``data: http://username:password@localhost:9200/ne_110m_populated_places_simple``
 
@@ -421,7 +421,7 @@ To publish an OpenSearch index, the following are required in your index:
 
 .. note::
 
-   For OpenSearch indexes that are password protect, a RFC1738 URL can be used as follows:
+   For OpenSearch indexes that are password protected, a RFC1738 URL can be used as follows:
 
    ``data: http://username:password@localhost:9200/ne_110m_populated_places_simple``
 

docs/source/publishing/ogcapi-processes.rst

@@ -96,6 +96,34 @@ manages job execution.  The manager concept is implemented as part of the pygeoa
 based on `TinyDB`_ for simplicity.  Custom manager plugins can be developed for more
 advanced job management capabilities (e.g. Kubernetes, databases, etc.).
 
+Elasticsearch
+^^^^^^^^^^^^^
+
+To publish an Elasticsearch index, the following are required:
+
+* indexes must be documents of jobs
+
+.. code-block:: yaml
+
+   server:
+       manager:
+           name: Elasticsearch
+           connection: http://localhost:9200/test-manager
+
+.. note::
+
+   Job results are expected to be stored in an ES job's ``location`` property.
+   A job document may set the ``location`` attribute to a string (instead of the default filepath),
+   in which case the plugin will render accordingly
+
+
+.. note::
+
+   For Elasticsearch indexes that are password protected, a RFC1738 URL can be used as follows:
+
+   ``data: http://username:password@localhost:9200/test-manager``
+
+
 TinyDB
 ^^^^^^
 

pygeoapi/plugin.py

@@ -82,6 +82,7 @@
     },
     'process_manager': {
         'Dummy': 'pygeoapi.process.manager.dummy.DummyManager',
+        'Elasticsearch': 'pygeoapi.process.manager.elasticsearch_.ElasticsearchManager',  # noqa
         'MongoDB': 'pygeoapi.process.manager.mongodb_.MongoDBManager',
         'TinyDB': 'pygeoapi.process.manager.tinydb_.TinyDBManager',
         'PostgreSQL': 'pygeoapi.process.manager.postgresql.PostgreSQLManager'

pygeoapi/process/manager/elasticsearch_.py

@@ -0,0 +1,251 @@
+# =================================================================
+#
+# Authors: Tom Kralidis <tomkralidis@gmail.com>
+#
+# Copyright (c) 2026 Tom Kralidis
+#
+# Permission is hereby granted, free of charge, to any person
+# obtaining a copy of this software and associated documentation
+# files (the "Software"), to deal in the Software without
+# restriction, including without limitation the rights to use,
+# copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the
+# Software is furnished to do so, subject to the following
+# conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+# OTHER DEALINGS IN THE SOFTWARE.
+#
+# =================================================================
+
+import json
+import logging
+from pathlib import Path
+from typing import Any, Tuple
+
+from elasticsearch import Elasticsearch, exceptions
+
+from pygeoapi.formats import FORMAT_TYPES, F_JSON, F_JSONLD
+from pygeoapi.process.base import (
+    JobNotFoundError,
+    JobResultNotFoundError,
+    ProcessorGenericError
+)
+from pygeoapi.process.manager.base import BaseManager
+from pygeoapi.util import JobStatus
+
+LOGGER = logging.getLogger(__name__)
+
+
+class ElasticsearchManager(BaseManager):
+    """Elasticsearch Manager"""
+
+    def __init__(self, manager_def: dict):
+        """
+        Initialize object
+
+        :param manager_def: manager definition
+
+        :returns: `pygeoapi.process.manager.elasticsearch_.ElasticsearchManager`  # noqa
+        """
+
+        super().__init__(manager_def)
+
+        self.is_async = True
+        self.supports_subscribing = True
+
+        self.es_host, self.index_name = self.connection.rsplit('/', 1)
+
+        LOGGER.debug('Setting Elasticsearch properties')
+
+        LOGGER.debug(f'host: {self.es_host}')
+        LOGGER.debug(f'index: {self.index_name}')
+
+        LOGGER.debug('Connecting to Elasticsearch')
+        self.es = Elasticsearch(self.es_host)
+        if not self.es.ping():
+            msg = f'Cannot connect to Elasticsearch: {self.es_host}'
+            LOGGER.error(msg)
+            raise ProcessorGenericError(msg)
+
+    def destroy(self) -> bool:
+        """
+        Destroy manager
+
+        :returns: `bool` status of result
+        """
+
+        LOGGER.debug(f'Destroying ES index {self.index_name}')
+        response = self.es.options(ignore_status=[400, 404]).indices.delete(
+            index=self.index_name)
+        LOGGER.debug(f'Result: {response}')
+
+        return True
+
+    def get_jobs(self, status: JobStatus = None, limit=None, offset=None
+                 ) -> dict:
+        """
+        Get jobs
+
+        :param status: job status (accepted, running, successful,
+                       failed, results) (default is all)
+        :param limit: number of jobs to return
+        :param offset: pagination offset
+
+        :returns: dict of list of jobs (identifier, status, process identifier)
+                  and numberMatched
+        """
+
+        jobs_list = []
+        query = {}
+
+        if status is not None:
+            LOGGER.debug('Applying status query')
+            query = {
+                'match': {
+                    'status': {
+                        'query': status
+                    }
+                }
+            }
+
+        es_results = self.es.search(index=self.index_name,
+                                    from_=offset, size=limit, **query)
+        matched = es_results['hits']['total']['value']
+        returned = len(es_results['hits']['hits'])
+
+        for hit in es_results['hits']['hits']:
+            job = hit['_source']
+            jobs_list.append(job)
+
+        return {
+            'jobs': jobs_list,
+            'numberMatched': matched,
+            'numberReturned': returned
+        }
+
+    def add_job(self, job_metadata: dict) -> str:
+        """
+        Add a job
+
+        :param job_metadata: `dict` of job metadata
+
+        :returns: identifier of added job
+        """
+
+        job_id = job_metadata['identifier']
+
+        LOGGER.debug(f"Inserting job with identifier {job_id}")
+        _ = self.es.index(index=self.index_name, id=job_id,
+                          body=job_metadata)
+        LOGGER.debug('Job added')
+
+        return job_id
+
+    def update_job(self, job_id: str, update_dict: dict) -> bool:
+        """
+        Updates a job
+
+        :param job_id: job identifier
+        :param update_dict: `dict` of property updates
+
+        :returns: `bool` of status result
+        """
+
+        LOGGER.debug(f'Updating job {job_id}')
+        _ = self.es.update(index=self.index_name, id=job_id, doc=update_dict)
+
+        return True
+
+    def delete_job(self, job_id: str) -> bool:
+        """
+        Deletes a job
+
+        :param job_id: job identifier
+
+        :raises JobNotFoundError: if the job_id does not correspond to a
+                                  known job
+        :return `bool` of status result
+        """
+
+        LOGGER.debug(f'Deleting job {job_id}')
+        _ = self.es.delete(index=self.index_name, id=job_id)
+
+        return True
+
+    def get_job(self, job_id: str) -> dict:
+        """
+        Get a single job
+
+        :param job_id: job identifier
+
+        :raises JobNotFoundError: if the job_id does not correspond to a
+                                  known job
+        :returns: `dict`  # `pygeoapi.process.manager.Job`
+        """
+
+        LOGGER.debug(f'Fetching job {job_id}')
+        try:
+            result = self.es.get(index=self.index_name, id=job_id)
+            result = result['_source']
+            return result
+        except exceptions.NotFoundError:
+            raise JobNotFoundError()
+
+    def get_job_result(self, job_id: str) -> Tuple[str, Any]:
+        """
+        Get a job's status, and actual output of executing the process
+
+        :param job_id: job identifier
+
+        :raises JobNotFoundError: if the job_id does not correspond to a
+                                  known job
+        :raises JobResultNotFoundError: if the job-related result cannot
+                                        be returned
+        :returns: `tuple` of mimetype and raw output
+        """
+
+        job_result = self.get_job(job_id)
+        location = job_result.get('location')
+        mimetype = job_result.get('mimetype')
+        job_status = JobStatus[job_result['status']]
+
+        if not job_status == JobStatus.successful:
+            # Job is incomplete
+            return (None,)
+        if location is None:
+            LOGGER.warning(f'job {job_id!r} -  unknown result location')
+            raise JobResultNotFoundError()
+        else:
+            try:
+                if Path(location).exists():
+                    location = Path(location)
+                    if mimetype in (None, FORMAT_TYPES[F_JSON],
+                                    FORMAT_TYPES[F_JSONLD]):
+                        with location.open('r', encoding='utf-8') as fh:
+                            result = json.load(fh)
+                    else:
+                        with location.open('rb') as filehandler:
+                            result = filehandler.read()
+                else:
+                    if mimetype in (None, FORMAT_TYPES[F_JSON],
+                                    FORMAT_TYPES[F_JSONLD]):
+                        print("QQQQ", location)
+                        print("QQQQ", type(location))
+                        result = json.loads(location)
+            except (TypeError, FileNotFoundError, json.JSONDecodeError):
+                raise JobResultNotFoundError()
+
+            return mimetype, result
+
+    def __repr__(self):
+        return f'<ElasticsearchManager> {self.name}'

pygeoapi/process/manager/tinydb_.py

@@ -40,7 +40,7 @@
 from pygeoapi.formats import FORMAT_TYPES, F_JSON, F_JSONLD
 from pygeoapi.process.base import (
     JobNotFoundError,
-    JobResultNotFoundError,
+    JobResultNotFoundError
 )
 from pygeoapi.process.manager.base import BaseManager
 from pygeoapi.util import JobStatus
@@ -57,7 +57,7 @@ def __init__(self, manager_def: dict):
 
         :param manager_def: manager definition
 
-        :returns: `pygeoapi.process.manager.base.BaseManager`
+        :returns: `pygeoapi.process.manager.tinydb_.TinyDBManager`
         """
 
         super().__init__(manager_def)

tests/data/test-manager.json

@@ -0,0 +1,46 @@
+{
+    "jobs": [
+        {
+            "type": "process",
+            "identifier": "cb23ae5b-c335-11f1-bb5c-d9d1498a46ce",
+            "process_id": "pywmdr-wmdr2-migrate",
+            "created": "2026-10-08T16:32:03.399740Z",
+            "started": "2026-10-08T16:32:03.399760Z",
+            "updated": "2026-10-08T16:32:03.404995Z",
+            "finished": "2026-10-08T16:32:03.404979Z",
+            "status": "failed",
+            "location": null,
+            "mimetype": "application/octet-stream",
+            "message": "InvalidParameterValue: Error executing process",
+            "progress": 5
+        },
+        {
+            "type": "process",
+            "identifier": "335fdbaf-c64c-11f1-ac1e-85ca8be9387b",
+            "process_id": "pywcmp-wis2-wcmp2-ets",
+            "created": "2026-10-12T14:50:00.693224Z",
+            "started": "2026-10-12T14:50:00.693242Z",
+            "updated": "2026-10-12T14:50:00.725403Z",
+            "finished": "2026-10-12T14:50:00.725386Z",
+            "status": "successful",
+            "location": "{\"foo\": \"bar\"}",
+            "mimetype": "application/json",
+            "message": "Job complete",
+            "progress": 100
+        },
+        {
+            "type": "process",
+            "identifier": "55555",
+            "process_id": "pywcmp-wis2-wcmp2-ets",
+            "created": "2026-10-12T14:50:00.693224Z",
+            "started": "2026-10-12T14:50:00.693242Z",
+            "updated": "2026-10-12T14:50:00.725403Z",
+            "finished": "2026-10-12T14:50:00.725386Z",
+            "status": "successful",
+            "location": null,
+            "mimetype": "application/json",
+            "message": "Job complete",
+            "progress": 100
+        }
+    ]
+}