add support for OGC API Publish-Subscribe Workflow - Part 1: Core (#2146) (#2220)

* add support for OGC API Publish-Subscribe Workflow - Part 1: Core (#2146)

* update docs

* update docs

* fix API deletion on not found items in backend

* fix ref

* fix docs

* add AsyncAPI support

* fix tests

* add exception handling

* update docs

Author: tomkralidis

.github/workflows/main.yml

@@ -126,6 +126,7 @@ jobs:
         pip3 install -r requirements-provider.txt
         pip3 install -r requirements-manager.txt
         pip3 install -r requirements-django.txt
+        pip3 install -r requirements-pubsub.txt
         pip3 install .
         pip3 install GDAL==`gdal-config --version`
     - name: setup test data ⚙️

docker/entrypoint.sh

@@ -46,6 +46,9 @@ fi
 if [[ -z "$PYGEOAPI_OPENAPI" ]]; then
     export PYGEOAPI_OPENAPI="${PYGEOAPI_HOME}/local.openapi.yml"
 fi
+if [[ -z "$PYGEOAPI_ASYNCAPI" ]]; then
+    export PYGEOAPI_ASYNCAPI="${PYGEOAPI_HOME}/local.asyncapi.yml"
+fi
 
 # gunicorn env settings with defaults
 SCRIPT_NAME=${SCRIPT_NAME:=/}
@@ -87,6 +90,12 @@ echo "Trying to generate openapi.yml"
 
 echo "openapi.yml generated continue to pygeoapi"
 
+echo "Trying to generate asyncapi.yml"
+/venv/bin/pygeoapi asyncapi generate ${PYGEOAPI_CONFIG} --output-file ${PYGEOAPI_ASYNCAPI}
+
+[[ $? -ne 0 ]] && error "asyncapi.yml could not be generated ERROR"
+echo "asyncapi.yml generated continue to pygeoapi"
+
 start_gunicorn() {
     # SCRIPT_NAME should not have value '/'
     [[ "${SCRIPT_NAME}" = '/' ]] && export SCRIPT_NAME="" && echo "make SCRIPT_NAME empty from /"

docs/source/administration.rst

@@ -24,12 +24,6 @@ To generate the OpenAPI document, run the following:
 
 This will dump the OpenAPI document as YAML to your system's ``stdout``.  To save to a file on disk, run:
 
-.. code-block:: bash
-
-   pygeoapi openapi generate /path/to/my-pygeoapi-config.yml > /path/to/my-pygeoapi-openapi.yml
-
-You can also write to a file explicitly via the ``--output-file`` option:
-
 .. code-block:: bash
 
    pygeoapi openapi generate /path/to/my-pygeoapi-config.yml --output-file /path/to/my-pygeoapi-openapi.yml
@@ -38,7 +32,7 @@ To generate the OpenAPI document as JSON, run:
 
 .. code-block:: bash
 
-   pygeoapi openapi generate /path/to/my-pygeoapi-config.yml -f json > /path/to/my-pygeoapi-openapi.json
+   pygeoapi openapi generate /path/to/my-pygeoapi-config.yml --format json --output-file /path/to/my-pygeoapi-openapi.yml
 
 .. note::
    Generate as YAML or JSON?  If your OpenAPI YAML definition is slow to render as JSON,
@@ -83,6 +77,11 @@ In UNIX:
     # or if OpenAPI JSON
     export PYGEOAPI_OPENAPI=/path/to/my-pygeoapi-openapi.json
 
+    # if your server supports AsyncAPI and Pub/Sub
+    export PYGEOAPI_ASYNCAPI=/path/to/my-pygeoapi-asyncapi.yml
+    # or if AsyncAPI JSON
+    export PYGEOAPI_ASYNCAPI=/path/to/my-pygeoapi-asyncapi.json
+
 In Windows:
 
 .. code-block:: bat
@@ -92,6 +91,14 @@ In Windows:
     # or if OpenAPI JSON
     set PYGEOAPI_OPENAPI=/path/to/my-pygeoapi-openapi.json
 
+    # if your server supports AsyncAPI and Pub/Sub
+    set PYGEOAPI_ASYNCAPI=/path/to/my-pygeoapi-asyncapi.yml
+    # or if AsyncAPI JSON
+    set PYGEOAPI_ASYNCAPI=/path/to/my-pygeoapi-asyncapi.json
+
+.. note::
+
+    More information on AsyncAPI and Pub/Sub can be found at :ref:`pubsub`.
 
 Summary
 -------

docs/source/configuration.rst

@@ -14,6 +14,7 @@ file whatever you wish; typical filenames end with ``.yml``.
 pygeoapi configuration contains the following core sections:
 
 - ``server``: server-wide settings
+- ``pubsub``: Publish-Subscribe settings (optional)
 - ``logging``: logging configuration
 - ``metadata``: server-wide metadata (contact, licensing, etc.)
 - ``resources``: dataset collections, processes and stac-collections offered by the server
@@ -90,6 +91,23 @@ For more information related to API design rules (the ``api_rules`` property in
         url_prefix: 'v{api_major}'  # adds a /v1 prefix to all URL paths
         version_header: X-API-Version  # add a response header of this name with the API version
 
+``pubsub``
+^^^^^^^^^^
+
+The ``pubsub`` section provides directives for enabling publication of CloudEvent messaages on item-based transactions
+
+
+.. code-block:: yaml
+
+  pubsub:
+      name: MQTT
+      broker:
+          url: mqtt://localhost:1883
+          channel: my/service/topic
+
+.. seealso::
+   :ref:`pubsub` for more information on Publish-Subscribe capabilities
+
 
 ``logging``
 ^^^^^^^^^^^

docs/source/index.rst

@@ -39,6 +39,7 @@ reference documentation on all aspects of the project.
    openapi
    publishing/index
    transactions
+   pubsub
    admin-api
    security
    plugins

docs/source/pubsub.rst

@@ -0,0 +1,135 @@
+.. _pubsub:
+
+Publish-Subscribe integration (Pub/Sub)
+=======================================
+
+pygeoapi supports Publish-Subscribe (Pub/Sub) integration by implementing
+the `OGC API Publish-Subscribe Workflow - Part 1: Core`_ (draft) specification.
+
+Pub/Sub integration can be enabled by defining a broker that pycsw can use to
+publish notifications on given topics using CloudEvents (as per the specification).
+
+When enabled, core functionality of Pub/Sub includes:
+
+- providing an AsyncAPI document (JSON and HTML)
+- providing the following links on the OGC API landing page:
+
+  - the broker link (``rel=hub`` link relation)
+  - the AsyncAPI JSON link (``rel=service-desc`` link relation and ``type=application/asyncapi+json`` media type)
+  - the AsyncAPI HTML link (``rel=service-doc`` link relation and ``type=text/html`` media type)
+
+- sending a notification message on the following events:
+
+  - feature or record transactions (create, replace, update, delete)
+  - process executions/job creation
+
+AsyncAPI
+--------
+
+`AsyncAPI`_ is the event-driven equivalent to :ref:`openapi`
+
+The official AsyncAPI specification can be found on the `AsyncAPI`_ website.  pygeoapi supports AsyncAPI version 3.0.0.
+
+AsyncAPI is an optional capability in pygeoapi.  To enable AsyncAPI, the following steps are required:
+
+- defining a ``pubsub`` section in configuration (see :ref:`configuration` and :ref:`brokers` for more information)
+- generating an AsyncAPI document
+- setting the ``PYGEOAPI_ASYNCAPI`` environment variable
+
+Creating the AsyncAPI document
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The AsyncAPI document is a YAML or JSON configuration which is generated from the pygeoapi configuration, and describes the server information, channels and the message payloads structures.
+
+To generate the AsyncAPI document, run the following:
+
+.. code-block:: bash
+
+   pygeoapi asyncapi generate /path/to/my-pygeoapi-config.yml
+
+This will dump the AsyncAPI document as YAML to your system's ``stdout``.  To save to a file on disk, run:
+
+.. code-block:: bash
+
+   pygeoapi asyncapi generate /path/to/my-pygeoapi-config.yml --output-file /path/to/my-pygeoapi-asyncapi.yml
+
+To generate the AsyncAPI document as JSON, run:
+
+.. code-block:: bash
+
+   pygeoapi asyncapi generate /path/to/my-pygeoapi-config.yml --format json --output-file /path/to/my-pygeoapi-asyncapi.json
+
+.. note::
+   Generate as YAML or JSON?  If your AsyncAPI YAML definition is slow to render as JSON,
+   saving as JSON to disk will help with performance at run-time.
+
+.. note::
+   The AsyncAPI document provides detailed information on query parameters, and dataset
+   property names and their data types.  Whenever you make changes to your pygeoapi configuration,
+   always refresh the accompanying AsyncAPI document.
+
+Validating the AsyncAPI document
+--------------------------------
+
+To ensure your AsyncAPI document is valid, pygeoapi provides a validation
+utility that can be run as follows:
+
+.. code-block:: bash
+
+   pygeoapi asyncapi validate /path/to/my-pygeoapi-asyncapi.yml
+
+.. _brokers:
+
+Brokers
+-------
+
+The following protocols are supported:
+
+MQTT
+^^^^
+
+Example directive:
+
+.. code-block:: yaml
+
+   pubsub:
+       name: MQTT
+       broker:
+           url: mqtt://localhost:1883
+           channel: messages/a/data  # optional
+           hidden: false # default
+
+HTTP
+^^^^
+
+Example directive:
+
+.. code-block:: yaml
+
+   pubsub:
+       name: HTTP
+       broker:
+           url: https://ntfy.sh
+           channel: messages-a-data  # optional
+           hidden: true  # default false
+
+.. note::
+
+   For any Pub/Sub endpoints requiring authentication, encode the ``url`` value as follows:
+
+   * ``mqtt://username:password@localhost:1883``
+   * ``https://username:password@localhost``
+
+   As with any section of the pygeoapi configuration, environment variables may be used as needed, for example
+   to set username/password information in a URL.  If ``pubsub.broker.url`` contains authentication, and
+   ``pubsub.broker.hidden`` is ``false``, the authentication information will be stripped from the URL
+   before displaying it on the landing page.
+
+.. note::
+
+   If a ``channel`` is defined, it is used as a prefix to the relevant OGC API endpoint used.
+
+   If a ``channel`` is not defined, only the relevant OGC API endpoint is used.
+
+.. _`OGC API Publish-Subscribe Workflow - Part 1: Core`: https://docs.ogc.org/DRAFTS/25-030.html
+.. _`AsyncAPI`: https://www.asyncapi.com

locale/en/LC_MESSAGES/messages.po

@@ -736,3 +736,21 @@ msgstr ""
 
 msgid "Instances"
 msgstr ""
+
+msgid "Pub/Sub Notifications"
+msgstr ""
+
+msgid "Pub/Sub broker"
+msgstr ""
+
+msgid "Subscribe to notifications from this service"
+msgstr ""
+
+msgid "AsyncAPI Definition"
+msgstr ""
+
+msgid "The AsyncAPI document as HTML"
+msgstr ""
+
+msgid "The AsyncAPI document as JSON"
+msgstr ""

pygeoapi/__init__.py

@@ -3,7 +3,7 @@
 # Authors: Tom Kralidis <tomkralidis@gmail.com>
 #          Ricardo Garcia Silva <ricardo.garcia.silva@geobeyond.it>
 #
-# Copyright (c) 2021 Tom Kralidis
+# Copyright (c) 2026 Tom Kralidis
 # Copyright (c) 2023 Ricardo Garcia Silva
 # Copyright (c) 2025 Angelos Tzotsos
 #
@@ -38,6 +38,8 @@
     from importlib.metadata import entry_points
 except ImportError:
     from importlib_metadata import entry_points
+
+from pygeoapi.asyncapi import asyncapi
 from pygeoapi.config import config
 from pygeoapi.openapi import openapi
 
@@ -110,3 +112,4 @@ def serve(ctx, server):
 
 cli.add_command(config)
 cli.add_command(openapi)
+cli.add_command(asyncapi)