refactoring

Author: ricardogsilva

docs/docker-example.md

@@ -5,9 +5,19 @@
 Prefect is able to run flows inside ephemeral docker containers.
 This requires that a Prefect work pool be created with the `docker` type:
 
-```shell
-prefect work-pool create --type docker my-pool
-```
+=== "uv"
+
+    ```shell
+    uv run prefect work-pool create --type docker my-pool
+    ```
+
+=== "pip"
+
+    ```shell
+    source .venv/bin/activate
+    
+    prefect work-pool create --type docker my-pool
+    ```
 
 Start a worker that consumes work from this newly-created pool
 
@@ -20,19 +30,28 @@ Start a worker that consumes work from this newly-created pool
 
     === "uv"
 
+        ```shell
+        uv add prefect-docker
+        ```
+
+    === "pip"
+
+        ```shell
+        pip install prefect-docker
+        ```
+
+=== "uv"
+
     ```shell
-    uv add prefect-docker
+    uv run prefect worker start --pool my-pool
     ```
 
-    === "pip"
+=== "pip"
 
     ```shell
-    pip install prefect-docker
+    prefect worker start --pool my-pool
     ```
 
-```shell
-prefect worker start --pool my-pool
-```
 
 #### Flow deployment
 
@@ -76,9 +95,17 @@ if __name__ == "__main__":
 
 Deploy it:
 
-```shell
-python simple_flow.py
-```
+=== "uv"
+
+    ```shell
+    uv run python simple_flow.py
+    ```
+
+=== "pip"
+
+    ```shell
+    python simple_flow.py
+    ```
 
 This creates a docker image named `pygeoapi-prefect/flows/simple-flow:{date}` with the flow contents and registers
 a deployment named `simple-flow/first-deployment` with the Prefect server. Because our `simple_flow.deploy()` call
@@ -87,9 +114,17 @@ includes `push=False`, this docker image lives in the local filesystem only.
 You can check the Prefect server UI in order to verify that your deployment is now registered. You can also used the
 Prefect API:
 
-```shell
-prefect deployment ls
-```
+=== "uv"
+
+    ```shell
+    uv run prefect deployment ls
+    ```
+
+=== "pip"
+
+    ```shell
+    prefect deployment ls
+    ```
 
 
 #### Pygeoapi processor configuration

docs/user-guide.md

@@ -151,15 +151,64 @@ either sync or async, as both types are supported by the Prefect manager.
 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`
+In order to be runnable via pygeoapi, your flows need to:
+
+1.  Implement the [protocol](https://typing.python.org/en/latest/spec/protocol.html)
+    defined in `pygeoapi_prefect.PygeoapiPrefectFlowProtocol`:
+
+    ```python
+    from typing import Protocol
+
+    class PygeoapiPrefectFlowProtocol(Protocol):
+        def __call__(
+            self,
+            processor_id: str,
+            pygeoapi_job_id: str,
+            inputs: dict,
+            outputs: dict | None = None,
+        ) -> None: ...
+    ```
+  
+2.  The flow must have at least one Prefect task, which is where the processing output is be generated. This task 
+    must return the generated output, which enables it to be managed by 
+    [Prefect's result management facilities](https://docs.prefect.io/v3/advanced/results). This means this 
+    result-generating task must:
+
+    -   Be configured to persist results;
+    -   Be configured with a result storage key that uses the pygeoapi job id - this is needed in order to ensure 
+        pygeoapi is able to reconstruct the result's storage key for retrieval;
+    -   The generated output must be a two-element tuple where the first element is the media type and the second 
+        element is the actual output.
+
+3.  The flow must already have been deployed in your Prefect environment. You can use any of the Prefect deployment 
+    types (local processes, docker containers, k8s, etc.). Check the [examples](docker-example.md) section for more information
+
+4.  Your pygeoapi configuration for the process needs to include a `prefect` section, with `deployment` and `metadata` 
+    sub-sections.
+
+This means that a very minimal flow looks like this:
 
 ```python
-from prefect import flow
+from prefect import flow, task
 
 @flow()
-def my_custom_flow()
+def my_custom_flow(
+    processor_id: str, 
+    pygeoapi_job_id: str, 
+    inputs: dict,
+    outputs: dict | None = None
+) -> None:
+     # perform whatever preparatory steps
+     generate_result(pygeoapi_job_id)
+
+
+@task(
+    persist_result=True,
+    result_storage_key="{parameters[pygeoapi_job_id].pickle}",
+    log_prints=True
+)
+def generate_result(pygeoapi_job_id: str) -> tuple[str, bytes]:
+    return "text/plain", "Hi there, I am a result".encode()
 
 ```
 

example-config.yml

@@ -77,11 +77,13 @@ resources:
   simple-flow:
     type: process
     processor:
+      job_control_options:
+        - sync-execute
+        - async-execute
       prefect:
         deployment:
           name: simple-flow/first-deployment
-          # result_storage_block: remote-file-system/test-sb1
-          result_storage_key_template: "{parameters[pygeoapi_job_id]}.pickle"
+          result_storage_key_template: "{job_id}.pickle"
         metadata:
           version: 0.0.1
           title: Hi world prefect example

src/pygeoapi_prefect/__init__.py

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

src/pygeoapi_prefect/cli.py

@@ -7,12 +7,10 @@
 from pygeoapi.process.manager.base import get_manager
 from pygeoapi.util import yaml_load
 
-from . import (
-    BasePrefectProcessor,
-    PrefectManager,
-    vanilla_flow,
-)
+from .process import PrefectDeploymentProcessor
+from .manager import PrefectManager
 from .schemas import ProcessId
+from . import vanilla_flow
 
 logger = logging.getLogger(__name__)
 
@@ -38,7 +36,7 @@ def deploy_processors_locally(
     to_serve = []
     for processor_id in manager.processes:
         match processor := manager.get_processor(ProcessId(processor_id)):
-            case BasePrefectProcessor():
+            case PrefectDeploymentProcessor():
                 logger.debug(
                     f"Skipping process {processor_id!r} - it provides its own "
                     f"deployment configuration"

src/pygeoapi_prefect/examples/simple_flow.py

@@ -0,0 +1,38 @@
+from prefect import flow, task
+
+
+@flow(log_prints=True)
+def simple_flow(
+        processor_id: str,
+        pygeoapi_job_id: str,  # noqa, this is used for naming flow_runs
+        inputs: dict,
+        outputs: dict | None = None,
+) -> None:
+    print(f"Hi from simple_flow locals: {locals()}")
+    generate_greeting(
+        name=inputs["name"],
+        pygeoapi_job_id=pygeoapi_job_id,
+        message=inputs.get("message"),
+    )
+
+
+@task(
+    persist_result=True,
+    result_storage_key="{parameters[pygeoapi_job_id]}.pickle",
+    log_prints=True,
+)
+def generate_greeting(
+        name: str,
+        pygeoapi_job_id: str,  # noqa, this is used for naming flow_runs
+        message: str | None = None
+) -> tuple[str, bytes]:
+    result = f"Hi {name}!"
+    if message:
+        result += f" {message}"
+    return "text/plain", result.encode()
+
+
+if __name__ == "__main__":
+    simple_flow.serve(
+        name="first-deployment",
+    )