adapting code to match current pygeoapi

Author: ricardogsilva

docs/development.md

@@ -45,10 +45,12 @@ uv run prefect server start
 The prefect UI shall now be available at http://localhost:4200
 
 
-If you want to deploy a pygeoapi process locally with prefect, then also start a prefect agent:
+If you want to deploy a pygeoapi process locally with prefect, then also start a prefect worker. Here we specify a
+worker of type `process` (i.e. flow runs execute locally, by spawning aditional Python processes) which consumes from
+a pool named `pygeoapi`:
 
 ```shell
-uv run prefect agent start --type process --pool pygeoapi
+uv run prefect worker start --type process --pool pygeoapi
 ```
 
 Now stand up pygeoapi

src/pygeoapi_prefect/__init__.py

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

src/pygeoapi_prefect/cli.py

@@ -1,14 +1,23 @@
+import logging
 from pathlib import Path
 
 import click
-from prefect.blocks.core import Block
-from prefect.deployments import Deployment
+from prefect import serve
 from prefect.filesystems import RemoteFileSystem
-from pygeoapi.process.manager import get_manager
-from pygeoapi.process import exceptions
+from pygeoapi.process.manager.base import get_manager
 from pygeoapi.util import yaml_load
 
-from .process import BasePrefectProcessor
+from . import (
+    BasePrefectProcessor,
+    PrefectManager,
+)
+from .manager import ProcessId
+from .vanilla_flow import (
+    generate_flow_run_name,
+    run_vanilla_processor,
+)
+
+logger = logging.getLogger(__name__)
 
 
 @click.group(name="prefect")
@@ -56,44 +65,104 @@ def create_storage_block(
     print("Done!")
 
 
-@root.command()
-@click.argument("process_id")
+@root.command(name="deploy-static")
 @click.option("-c", "--pygeoapi-config", type=Path, envvar="PYGEOAPI_CONFIG")
-def deploy_process(
-    process_id: str,
-    pygeoapi_config: Path,
+def deploy_processors_locally(
+        pygeoapi_config: Path,
 ):
-    """Create and apply prefect deployment for PROCESS_ID.
-
-    Configure deployment parameters for the process in pygeoapi's configuration
-    file.
-    """
+    """Deploy pygeoapi processes via Prefect, locally."""
     with pygeoapi_config.open() as fh:
         config = yaml_load(fh)
     manager = get_manager(config)
-    try:
-        processor = manager.get_processor(process_id)
-    except exceptions.UnknownProcessError as err:
-        raise click.BadParameter(f"Process {process_id!r} not found") from err
-    else:
-        if isinstance(processor, BasePrefectProcessor):
-            if processor.deployment_info is not None:
-                print(f"Deploying process {process_id!r} with prefect...")
-                if (sb := processor.deployment_info.storage_block) is not None:
-                    storage = Block.load(sb)
-                else:
-                    storage = None
-                deployment = Deployment.build_from_flow(
-                    processor.process_flow,
-                    name=processor.deployment_info.name,
-                    work_queue_name=processor.deployment_info.queue,
-                    storage=storage,
-                    path=processor.deployment_info.storage_sub_path,
-                    ignore_file=None,
+    if not isinstance(manager, PrefectManager):
+        raise SystemExit(
+            "Cannot deploy as prefect flows - pygeoapi_prefect.PrefectManager "
+            "is not being used as the pygeoapi process manager"
+        )
+    to_serve = []
+    for processor_id in manager.processes:
+        match processor := manager.get_processor(ProcessId(processor_id)):
+            case BasePrefectProcessor():
+                logger.debug(
+                    f"Skipping process {processor_id!r} - it provides its own "
+                    f"deployment configuration"
+                )
+                continue
+            case BasePrefectProcessor():
+                configured_flow = run_vanilla_processor.with_options(
+                    name=processor.metadata["id"],
+                    version=processor.metadata.get("version"),
+                    flow_run_name=generate_flow_run_name,
+                    validate_parameters=True,
+                )
+                flow_deployment = configured_flow.to_deployment(
+                    name=f"pygeoapi-{processor_id}-local",
+                    parameters={"processor_id": processor_id},
                 )
-                deployment.apply()
-                print("Done!")
-            else:
-                raise click.Abort("Deployment not specified in pygeoapi config file")
-        else:
-            print(f"Process {process_id!r} is not deployable with prefect, skipping...")
+                to_serve.append(flow_deployment)
+            case _:
+                logger.warning(f"Unknown processor type {processor_id}, ignoring...")
+                continue
+        # processor = manager.get_processor(ProcessId(processor_id))
+        # if isinstance(processor, BasePrefectProcessor):
+        #     logger.debug(
+        #         f"Skipping process {processor_id!r} - it provides its own "
+        #         f"deployment configuration"
+        #     )
+        #     continue
+        # else:
+        #     configured_flow = run_vanilla_processor.with_options(
+        #         name=processor.metadata["id"],
+        #         version=processor.metadata.get("version"),
+        #         flow_run_name=generate_flow_run_name,
+        #         validate_parameters=True,
+        #     )
+        #     flow_deployment = configured_flow.to_deployment(
+        #         name=f"pygeoapi-{processor_id}-local",
+        #         parameters={"processor_id": processor_id},
+        #     )
+        #     to_serve.append(flow_deployment)
+    serve(*to_serve)
+
+
+# @root.command()
+# @click.argument("process_id")
+# @click.option("-c", "--pygeoapi-config", type=Path, envvar="PYGEOAPI_CONFIG")
+# def deploy_process(
+#     process_id: str,
+#     pygeoapi_config: Path,
+# ):
+#     """Create and apply prefect deployment for PROCESS_ID.
+#
+#     Configure deployment parameters for the process in pygeoapi's configuration
+#     file.
+#     """
+#     with pygeoapi_config.open() as fh:
+#         config = yaml_load(fh)
+#     manager = get_manager(config)
+#     try:
+#         processor = manager.get_processor(process_id)
+#     except UnknownProcessError as err:
+#         raise click.BadParameter(f"Process {process_id!r} not found") from err
+#     else:
+#         if isinstance(processor, BasePrefectProcessor):
+#             if processor.deployment_info is not None:
+#                 print(f"Deploying process {process_id!r} with prefect...")
+#                 if (sb := processor.deployment_info.storage_block) is not None:
+#                     storage = Block.load(sb)
+#                 else:
+#                     storage = None
+#                 deployment = Deployment.build_from_flow(
+#                     processor.process_flow,
+#                     name=processor.deployment_info.name,
+#                     work_queue_name=processor.deployment_info.queue,
+#                     storage=storage,
+#                     path=processor.deployment_info.storage_sub_path,
+#                     ignore_file=None,
+#                 )
+#                 deployment.apply()
+#                 print("Done!")
+#             else:
+#                 raise click.Abort("Deployment not specified in pygeoapi config file")
+#         else:
+#             print(f"Process {process_id!r} is not deployable with prefect, skipping...")

src/pygeoapi_prefect/examples/hi_prefect_world.py

@@ -25,10 +25,7 @@
 from pygeoapi_prefect.process import BasePrefectProcessor
 
 
-@flow(
-    persist_result=True,
-    log_prints=True,
-)
+@flow(log_prints=True)
 def hi_prefect_world(
     job_id: str,
     result_storage_block: str | None,

src/pygeoapi_prefect/examples/simple_prefect.py

@@ -23,11 +23,7 @@
 from pygeoapi_prefect.process import BasePrefectProcessor
 
 
-# When defining a prefect flow that will be deployed by prefect to some
-# infrastructure, be sure to specify persist_result=True - otherwise the
-# pygeoapi process manager will not be able to work properly
 @flow(
-    persist_result=True,
     log_prints=True,
 )
 def simple_flow(

src/pygeoapi_prefect/manager.py

@@ -48,6 +48,7 @@
     exceptions,
     prefect_client,
     schemas,
+    vanilla_flow,
 )
 from .process import (
     BasePrefectProcessor,
@@ -399,33 +400,25 @@ def execute_process(
             subscriber=subscriber,
         )
         job_id = PygeoapiPrefectJobId(str(uuid.uuid4()))
-        # processor = self.get_processor(process_id)
-        match processor := self.get_processor(process_id):
-            case BasePrefectProcessor():
-                chosen_mode, headers = select_prefect_processor_execution_mode(
-                    execution_mode, processor)
-                internal_job_status = execute_prefect_processor(
-                    job_id, processor, chosen_mode, execution_request
-                )
-                return (
-                    job_id,
-                    None,
-                    None,
-                    internal_job_status.status,
-                    additional_headers
-                )
-            case BaseProcessor():
-                output_media_type, generated_output, current_job_status = (
-                    execute_vanilla_processor_sync(
-                        job_id, processor, execution_request, output_dir=self.output_dir)
-                )
-                return (
-                    job_id,
-                    output_media_type,
-                    generated_output,
-                    current_job_status,
-                    {"Preference-applied": RequestedProcessExecutionMode.wait.value},
-                )
+        chosen_mode, headers = select_execution_mode(execution_mode)
+        processor = self.get_processor(process_id)
+        # https://docs.prefect.io/v3/how-to-guides/deployments/run-deployments#run-a-deployment-from-python
+        if chosen_mode == ProcessExecutionMode.sync_execute:
+            # if it's a BaseProcessor type, run our locally deployed flow
+            # if it's a PrefectProcessor type, run the name of the deployment
+            # 1. get the name of the deployment
+            ...
+        else:
+            ...
+        match processor, chosen_mode:
+            case BasePrefectProcessor(), ProcessExecutionMode.sync_execute:
+                ...
+            case BasePrefectProcessor(), ProcessExecutionMode.async_execute:
+                ...
+            case BaseProcessor(), ProcessExecutionMode.sync_execute:
+                ...
+            case BaseProcessor(), ProcessExecutionMode.async_execute:
+                ...
             case _:
                 raise ProcessorExecuteError("Unknown processor type")
 
@@ -546,13 +539,19 @@ def get_job_status_from_flow_run(
         prefect_flow: Flow
 ) -> schemas.JobStatusInfoInternal:
     job_id = PygeoapiPrefectJobId.from_flow_run_name(flow_run.name)
+    logger.info(f"{flow_run=}")
+    logger.info(f"{flow_run.parameters=}")
+    logger.info(f"{job_id=}")
     try:
-        partial_info = flow_run.state.result()
-        generated_outputs = partial_info.generated_outputs
+        media_type, generated_output = flow_run.state.result()
+        logger.info(f"{media_type=}")
+        logger.info(f"{generated_output=}")
     except MissingResult as err:
         logger.warning(f"Could not get flow_run results: {err}")
-        generated_outputs = None
+        generated_output = None
     execution_request = schemas.ExecuteRequest(**flow_run.parameters["execution_request"])
+    logger.info(f"{execution_request=}")
+    logger.info(f"{flow_run.state_type=}")
     return schemas.JobStatusInfoInternal(
         jobID=job_id,
         status=PREFECT_STATE_MAP[flow_run.state_type],
@@ -562,10 +561,28 @@ def get_job_status_from_flow_run(
         finished=flow_run.end_time,
         requested_response_type=execution_request.response,
         requested_outputs=execution_request.outputs,
-        generated_outputs=generated_outputs,
+        generated_outputs=generated_output,
     )
 
 
+def select_execution_mode(
+        requested: RequestedProcessExecutionMode | None,
+) -> tuple[ProcessExecutionMode, dict[str, str]]:
+    """Select the execution mode to be employed for the input processor.
+
+    According to the OGC API - Processes (Section 7.11.2.3, Requirement 25)
+    the execution mode shall default to `sync` mode of omitted by the client
+    (and supported by the server). The Prefect process manager is able to
+    run in either sync or async modes, regardless of the processor.
+    """
+    headers = {"Preference-Applied": RequestedProcessExecutionMode.wait.value}
+    chosen = ProcessExecutionMode.sync_execute
+    if requested == RequestedProcessExecutionMode.respond_async:
+        chosen = ProcessExecutionMode.async_execute
+        headers["Preference-Applied"] = RequestedProcessExecutionMode.respond_async.value
+    return chosen, headers
+
+
 def select_prefect_processor_execution_mode(
         requested: RequestedProcessExecutionMode | None,
         processor: BasePrefectProcessor,
@@ -598,6 +615,7 @@ def execute_vanilla_processor_sync(
         processor: PygeoapiProcessorProtocol,
         execution_request: schemas.ExecuteRequest,
         output_dir: Path | None,
+        prefect_worker_pool: str | None = None,
 ) -> tuple[MediaType, Any, JobStatus]:
     """Execute a regular pygeoapi processor locally via prefect.
 

src/pygeoapi_prefect/vanilla_flow.py

@@ -0,0 +1,34 @@
+"""Run a pygeoapi processor as a Prefect flow.
+
+Note that this module cannot have any relative imports, as it is intended that
+it be importable by the Prefect worker as a script.
+"""
+import os
+from pathlib import Path
+from typing import Any
+
+from prefect import flow
+from prefect.runtime import flow_run
+from pygeoapi.process.manager.base import get_manager
+from pygeoapi.util import yaml_load
+
+
+def generate_flow_run_name():
+    pygeoapi_job_id = (flow_run.parameters or {}).get("pygeoapi_job_id") or "unknown"
+    return f"pygeoapi_job_{pygeoapi_job_id}"
+
+
+@flow(persist_result=True, log_prints=True)
+def run_vanilla_processor(
+        processor_id: str,
+        data_: dict,
+        pygeoapi_job_id: str,  # noqa, this is used for naming flow_runs
+        outputs: dict | None = None,
+) -> tuple[str, Any]:
+    config_path = Path(os.environ["PYGEOAPI_CONFIG"])
+    with config_path.open() as fh:
+        config = yaml_load(fh)
+    manager = get_manager(config)
+    processor = manager.get_processor(processor_id)
+    processor_output_media_type, processor_output = processor.execute(data_, outputs)
+    return processor_output_media_type, processor_output