allow running vanilla processes in-band, when called as sync

Author: ricardogsilva

docs/guide.md docs/user-guide.mddocs/guide.md renamed to docs/user-guide.md

@@ -1,3 +1,8 @@
+---
+hide:
+  - navigation
+---
+
 # User guide
 
 ## Installation
@@ -69,6 +74,25 @@ server:
     name: pygeoapi_prefect.PrefectManager
 ```
 
+#### Prefect manager configuration
+
+The Prefect Manager accepts the following configuration parameters, all of which are optional.
+These should be specified as properties of the `server.manager` object:
+
+-  `use_vanilla_processor_deployments: bool = True` - Whether to call native pygeoapi processors via the 
+   Prefect deployment interface. If this is enabled (which is the default):
+
+    - Jobs are executed via the Prefect worker, which you must start (as mentioned below)
+    - Jobs are coordinated by the Prefect scheduler, which means that they may not start immediately
+    - Jobs are run in a different process than pygeoapi
+    - This means that job execution is slower, as there is a temporal overhead which 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
+
+- `sync_job_execution_timeout_seconds: int = 60` - How much time to give a sync job to finish its 
+  processing before declaring it as failed
+
 
 #### Processors
 

mkdocs.yml

@@ -7,14 +7,14 @@ edit_uri: edit/main/docs/
 
 nav:
   - Home: 'index.md'
-  - New User Guide: 'guide.md'
-  - User Guide:
-      - Installation: 'user-guide/installation.md'
-      - Defining processes: 'user-guide/defining-processes.md'
-      - Running prefect: 'user-guide/prefect.md'
-      - Deployments: 'user-guide/deployments.md'
-      - Executing processes: 'user-guide/executing-processes.md'
-      - Complete example: 'user-guide/complete-example.md'
+  - User Guide: 'user-guide.md'
+  # - User Guide:
+    # - Installation: 'user-guide/installation.md'
+    # - Defining processes: 'user-guide/defining-processes.md'
+    # - Running prefect: 'user-guide/prefect.md'
+    # - Deployments: 'user-guide/deployments.md'
+    # - Executing processes: 'user-guide/executing-processes.md'
+    # - Complete example: 'user-guide/complete-example.md'
   - Development: 'development.md'
 
 theme:
@@ -27,6 +27,7 @@ theme:
     - content.tabs.link
     - header.autohide
     - navigation.footer
+    - navigation.indexes
     - navigation.tabs
     - navigation.tabs.sticky
 

src/pygeoapi_prefect/cli.py

@@ -45,12 +45,7 @@ def deploy_processors_locally(
                 )
                 continue
             case BaseProcessor():
-                configured_flow = vanilla_flow.run_vanilla_processor.with_options(
-                    name=processor.metadata["id"],
-                    version=processor.metadata.get("version"),
-                    flow_run_name=vanilla_flow.generate_flow_run_name,
-                    validate_parameters=True,
-                )
+                configured_flow = vanilla_flow.get_processor_as_flow(processor)
                 flow_deployment = configured_flow.to_deployment(
                     name=vanilla_flow.get_deployment_name(processor_id),
                     parameters={"processor_id": processor_id},

src/pygeoapi_prefect/manager.py

@@ -2,7 +2,6 @@
 
 import copy
 import importlib
-import json
 import logging
 import uuid
 from collections.abc import (
@@ -18,11 +17,9 @@
 import httpx
 import jsonschema.exceptions
 import jsonschema.validators
-from prefect import flow
 from prefect.blocks.core import Block
 from prefect.client.schemas import FlowRun
 from prefect.deployments import run_deployment
-from prefect.exceptions import MissingResult
 from prefect.results import (
     ResultRecord,
     ResultStore,
@@ -102,6 +99,7 @@ class PrefectManager:
     _processor_configurations: dict[ProcessId, dict[str, Any]]
 
     is_async: bool = True
+    use_vanilla_processor_deployments: bool
     supports_subscribing: bool
     prefect_state_map = {
         StateType.SCHEDULED: JobStatus.accepted,
@@ -120,6 +118,9 @@ class PrefectManager:
     sync_job_execution_timeout_seconds: int
 
     def __init__(self, manager_def: dict[str, Any]):
+        self.use_vanilla_processor_deployments = manager_def.get(
+            "use_vanilla_processor_deployments", True)
+        # self.use_vanilla_processor_deployments = False
         self.name = ".".join(
             (self.__class__.__module__, self.__class__.__qualname__)
         )
@@ -322,6 +323,15 @@ def execute_process(
         if execution_mode == RequestedProcessExecutionMode.respond_async:
             chosen_mode = ProcessExecutionMode.async_execute
             response_headers["Preference-Applied"] = RequestedProcessExecutionMode.respond_async.value
+        if isinstance(requested_outputs, Sequence) and not isinstance(requested_outputs, str):
+            outs = {out_name: ExecutionOutput() for out_name in requested_outputs}
+        elif isinstance(requested_outputs, Mapping):
+            outs = {
+                out_name: ExecutionOutput(**out_info)
+                for out_name, out_info in requested_outputs.items()
+            }
+        else:
+            outs = None
 
         match processor := self.get_processor(process_id):
             case BasePrefectProcessor():
@@ -334,15 +344,7 @@ def execute_process(
                 )
             case _:
                 raise ProcessError(f"Unknown processor type {type(processor)!r}")
-        if isinstance(requested_outputs, Sequence) and not isinstance(requested_outputs, str):
-            outs = {out_name: ExecutionOutput() for out_name in requested_outputs}
-        elif isinstance(requested_outputs, Mapping):
-            outs = {
-                out_name: ExecutionOutput(**out_info)
-                for out_name, out_info in requested_outputs.items()
-            }
-        else:
-            outs = None
+
         execution_request = ExecuteRequest(
             deployment_info=deployment_info,
             inputs=data_,
@@ -351,8 +353,15 @@ def execute_process(
             subscriber=subscriber,
         )
         if chosen_mode == ProcessExecutionMode.sync_execute:
-            media_type, generated_output = self._execute_job_sync(
-                job_id, processor, execution_request)
+            if isinstance(processor, BaseProcessor) and not self.use_vanilla_processor_deployments:
+                print(f"Executing processor {processor.metadata['id']!r} in-process...")
+                media_type, generated_output = _execute_job_sync_without_deployment(
+                    job_id, processor, execution_request
+                )
+            else:
+                print(f"Executing processor {processor.metadata['id']!r} via Prefect deployment...")
+                media_type, generated_output = self._execute_job_sync(
+                    job_id, processor, execution_request)
             print(f"{media_type=}")
             print(f"{generated_output=}")
             return job_id, media_type, generated_output, JobStatus.successful, response_headers
@@ -464,3 +473,29 @@ def _retrieve_result_from_prefect(
     media_type = MediaType(result_record.result[0])
     generated_output = result_record.result[1]
     return media_type, generated_output
+
+
+def _execute_job_sync_without_deployment(
+        job_id: PygeoapiPrefectJobId,
+        processor: BaseProcessor,
+        execution_request: ExecuteRequest,
+) -> tuple[MediaType, Any]:
+    """Execute a job in sync mode by calling a flow directly
+
+    This execution mode is only suitable for:
+
+    - flows that wrap vanilla pygeoapi processors
+    - flows whose source code is reachable in the same process
+      as this one
+    """
+    configured_flow = vanilla_flow.get_processor_as_flow(processor)
+    configured_flow(
+        processor.metadata["id"],
+        job_id,
+        execution_request.inputs,
+        execution_request.outputs
+    )
+    return _retrieve_result_from_prefect(
+        result_storage_block=execution_request.deployment_info.result_storage_block,
+        result_storage_key_template=execution_request.deployment_info.result_storage_key_template
+    )

src/pygeoapi_prefect/vanilla_flow.py

@@ -7,14 +7,25 @@
 from pathlib import Path
 
 from prefect import (
+    Flow,
     flow,
     task,
 )
 from prefect.runtime import flow_run
 from pygeoapi.process.manager.base import get_manager
+from pygeoapi.process.base import BaseProcessor
 from pygeoapi.util import yaml_load
 
 
+def get_processor_as_flow(processor: BaseProcessor) -> Flow:
+    return run_vanilla_processor.with_options(
+        name=processor.metadata["id"],
+        version=processor.metadata.get("version"),
+        flow_run_name=generate_flow_run_name,
+        validate_parameters=True,
+    )
+
+
 def get_deployment_name(processor_id: str) -> str:
     return f"pygeoapi-{processor_id}-local"