Improve docs

Author: ricardogsilva

.pre-commit-config.yaml

@@ -8,12 +8,12 @@ repos:
       - id: end-of-file-fixer
       - id: check-added-large-files
         args: ["--maxkb=1000"]
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.15.10
+    hooks:
+      - id: ruff-check
+      - id: ruff-format
   # the below hooks are disabled for now
-  # - repo: https://github.com/astral-sh/ruff-pre-commit
-  #   rev: v0.9.6
-  #   hooks:
-  #     - id: ruff
-  #     - id: ruff-format
   # - repo: local
   #   hooks:
   #     - id: ty

docs/user-guide.md

@@ -36,15 +36,27 @@ In order to work with pygeoapi-prefect, you need to set up:
 
 - a Prefect server;
 - pygeoapi;
-- at least one Prefect worker.
+- (if you want to support async execution) at least one Prefect worker.
 
 
 ### Prefect server
 
 After installation, you need to have a Prefect server running and to set some environment variables
 so that the Prefect Manager is able to communicate with it.
 
-As the most basic setup, just start a local Prefect server
+As the most basic setup:
+
+generate an env file for configuring Prefect, ensuring that you
+enable [persistence of results](https://docs.prefect.io/v3/advanced/results).
+
+```shell
+cat << EOF > prefect-env.env
+PREFECT_API_URL=http://127.0.0.1:4200/api
+PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
+EOF
+```
+
+Then start a local Prefect server
 
 === "uv"
 
@@ -79,6 +91,51 @@ server:
 The Prefect Manager accepts the following configuration parameters, all of which are optional.
 These should be specified as properties of the `server.manager` object:
 
+- `enable_sync_job_execution: bool = False` - Whether to accept execution requests that run synchronously.
+
+    !!! NOTE
+
+        This setting is disabled by default and we recommend keeping it disabled - Executing syncronously risks
+        overloading the server in case a large number of requestes is received.
+
+        If needed, the metadata of
+        each processor is allowed to override this setting, making it possible to selectively enable sync
+        execution for those processors in which it is a useful execution mode.
+
+        #### Controlling processor execution modes via pygeoapi config
+
+        By default, pygeoapi does not allow modying a processor's metadata in its configuration and assumes that
+        all parameters are to be set together with the processor code. This has the disadvantage of making it
+        impossible to toggle a processor's support for sync execution on and off without modifying the source code.
+        In order to ease configuration, `PrefectManager` will look for a `job_control_options` configuration key on
+        its own configuration file, which has the same meaning as the `jobControlOptions` metadata key, _i.e._ it
+        should be a list of strings with at least one entry and can contain `sync-execute` and `async-execute` entries.
+
+        In the following example we disable sync mode in the manager but re-enable it in the `hello-world`
+        processor configuration:
+
+        ```yaml
+        # pygeoapi configuration file
+        server:
+          manager:
+            name: pygeoapi_prefect.PrefectManager
+            enable_sync_job_execution: false
+        resources:
+          hello-world:
+            type: process
+            processor:
+              name: HelloWorld
+              job_control_options:
+                - sync-execute
+        ```
+
+
+- `enable_async_job_execution: bool = True` - Whether to accept execution requests that run asynchronously.
+  This is enabled by default and is the recommended way to operate. Async execution defers the scheduling of
+  processor execution to the Prefect scheduler, allowing for more stable operation, as the scheduler can control
+  when to execute jobs depending on the load. Similarly to the previous parameter, this can also be
+  overridden on a per-processor basis.
+
 -  `use_deployment_for_sync_requests: bool = False` - Whether to call native pygeoapi processors via the
    Prefect deployment interface or not. If this is enabled:
 
@@ -88,7 +145,11 @@ These should be specified as properties of the `server.manager` object:
     - This means that job execution is slower, as there is a temporal overhead that is introduced by
       the coordination that happens between the Prefect scheduler service and the Prefect worker
 
-    If this is disabled, then jobs are executed immediately and run in the same process as pygeoapi
+    If this is disabled (the default), then jobs are executed immediately and run in the same process as
+    pygeoapi.
+
+    Note that regardless of this being enabled or not, all pygeoapi processes execution requests are tracked by
+    Prefect and can be monitored using the Prefect UI.
 
 - `sync_job_execution_timeout_seconds: int = 60` - How much time to give a sync job to finish its
   processing before declaring it as failed
@@ -115,10 +176,62 @@ resources:
       name: HelloWorld
 ```
 
+The `PrefectManager` wraps these processors inside a Prefect [flow](https://docs.prefect.io/v3/concepts/flows) and
+is able to execute them by generating Prefect flow runs whenever an execution request is received. Execution can be
+either sync or async, as both types are supported by the Prefect manager.
+
+
+###### Sync execution
+
+Whenever pygeoapi receives a sync processor execution request, such as:
+
+```shell
+curl \
+    -X POST \
+    "http://localhost:5000/processes/hello-world/execution" \
+    -H "Content-Type: application/json" \
+    -d '{"inputs": {"name": "Joe"}}'
+```
+
+This is turned into a Prefect flow run and is executed locally, in the same process as pygeoapi. In other words,
+the pygeoapi processor is wrapped as a Prefect flow and is run by the Prefect manager, using regular function calls.
+
+Being managed by Prefect, this means.
+
+
+###### Async execution
+
+When pygeoapi receives an async processor execution request, such as:
+
+```shell
+curl \
+    -X POST \
+    "http://localhost:5000/processes/hello-world/execution" \
+    -H "Content-Type: application/json" \
+    -H "Prefer: respond-async" \
+    -d '{"inputs": {"name": "Joe"}}'
+```
+
+This is turned into a Prefect flow and is executed via the respective processor deployment
+
 
 ##### custom prefect-aware processes
 
-TBD
+If you prefer, you can write Prefect flows, deploy them using any of the multiple techniques supported by Prefect
+and then adapt them to run as pygeoapi processors.
+
+In order to do so, your Prefect flows need to implement the
+[protocol](https://typing.python.org/en/latest/spec/protocol.html)
+defined in `pygeoapi_prefect.PygeoapiPrefectFlowProtocol`
+
+```python
+from prefect import flow
+
+@flow()
+def my_custom_flow()
+
+```
+
 
 #### Launching pygeoapi
 

example-config.yml

@@ -18,8 +18,8 @@ server:
     url: https://tile.openstreetmap.org/{z}/{x}/{y}.png
     attribution: '&copy; <a href="https://openstreetmap.org/copyright">OpenStreetMap contributors</a>'
   manager:
-    name: pygeoapi_prefect.manager.PrefectManager
-    use_deployment_for_sync_requests: true
+    name: pygeoapi_prefect.PrefectManager
+    # use_deployment_for_sync_requests: false
     # name: TinyDB
     # connection: /tmp/pygeoapi-process-manager.db
     output_dir: /tmp/

pyproject.toml

@@ -35,6 +35,9 @@ prefect = "pygeoapi_prefect.cli:root"
 [tool.uv.sources]
 pygeoapi = { path = "../../pygeoapi", editable = true }
 
+[tool.ty.analysis]
+allowed-unresolved-imports = ["pygeoapi.**"]
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

src/pygeoapi_prefect/__init__.py

@@ -1,7 +1,5 @@
 from .manager import PrefectManager
 from .process import BasePrefectProcessor
+from .protocols import PygeoapiPrefectFlowProtocol
 
-__all__ = [
-    "BasePrefectProcessor",
-    "PrefectManager",
-]
+__all__ = ["BasePrefectProcessor", "PrefectManager", "PygeoapiPrefectFlowProtocol"]

src/pygeoapi_prefect/cli.py

@@ -24,7 +24,7 @@ def root(): ...
 @root.command(name="deploy-static")
 @click.option("-c", "--pygeoapi-config", type=Path, envvar="PYGEOAPI_CONFIG")
 def deploy_processors_locally(
-        pygeoapi_config: Path,
+    pygeoapi_config: Path,
 ):
     """Deploy pygeoapi processes via Prefect, locally."""
     with pygeoapi_config.open() as fh:

src/pygeoapi_prefect/examples/hi_prefect_world.py

@@ -1,7 +1,6 @@
 """Example pygeoapi process"""
 
 from prefect import (
-    Flow,
     flow,
     get_run_logger,
     task,
@@ -19,7 +18,6 @@
     OutputExecutionResultInternal,
     InternalProcessDescription,
     ProcessInput,
-    ProcessJobControlOption,
     ProcessOutput,
 )
 from pygeoapi_prefect.process import BasePrefectProcessor
@@ -84,8 +82,8 @@ def generate_status_info(
     job_id: str, process_id: str, result_path: str, result_media_type: str
 ):
     return JobStatusInfoInternal(
-        jobID=job_id,
-        processID=process_id,
+        job_id=job_id,
+        process_id=process_id,
         status=JobStatus.successful,
         generated_outputs={
             "result": OutputExecutionResultInternal(
@@ -107,19 +105,19 @@ class HiPrefectWorldProcessor(BasePrefectProcessor):
             "name": ProcessInput(
                 title="Name",
                 description="Some name you think is cool. It will be echoed back.",
-                schema={"type": "string"},
+                schema_={"type": "string"},
                 keywords=["cool-name"],
             ),
             "message": ProcessInput(
                 title="Message",
                 description="An optional additional message to be echoed to the world",
-                schema={"type": "string"},
-                minOccurs=0,
+                schema_={"type": "string"},
+                min_occurs=0,
             ),
         },
         outputs={
             "result": ProcessOutput(
-                schema={
+                schema_={
                     "type": "string",
                     "contentMediaType": "text/plain",
                 },

src/pygeoapi_prefect/examples/simple_prefect.py

@@ -16,7 +16,6 @@
     JobStatusInfoInternal,
     InternalProcessDescription,
     ProcessInput,
-    ProcessJobControlOption,
     ProcessOutput,
     OutputExecutionResultInternal,
 )
@@ -54,8 +53,8 @@ def simple_flow(
         result_path = f"{job_id}/output-result.txt"
         file_system.write_path(result_path, result_value.encode("utf-8"))
         return JobStatusInfoInternal(
-            jobID=job_id,
-            processID=process_description.id,
+            job_id=job_id,
+            process_id=process_description.id,
             status=JobStatus.successful,
             generated_outputs={
                 "result": OutputExecutionResultInternal(
@@ -81,19 +80,19 @@ class SimpleFlowProcessor(BasePrefectProcessor):
             "name": ProcessInput(
                 title="Name",
                 description="Some name you think is cool. It will be echoed back.",
-                schema={"type": "string"},
+                schema_={"type": "string"},
                 keywords=["cool-name"],
             ),
             "message": ProcessInput(
                 title="Message",
                 description="An optional additional message to be echoed to the world",
-                schema={"type": "string"},
-                minOccurs=0,
+                schema_={"type": "string"},
+                min_occurs=0,
             ),
         },
         outputs={
             "result": ProcessOutput(
-                schema={
+                schema_={
                     "type": "string",
                     "contentMediaType": "text/plain",
                 },

src/pygeoapi_prefect/exceptions.py

@@ -1,4 +1,4 @@
-class PygeoapiPrefectError(Exception):  ...
+class PygeoapiPrefectError(Exception): ...
 
 
 class InvalidProcessorDefinitionError(PygeoapiPrefectError): ...