Package reorganization, setup, and samples (#2)

Author: cgillum

.flake8

@@ -0,0 +1,6 @@
+[flake8]
+ignore = E501,C901
+exclude = 
+    .git
+    *_pb2*
+    __pycache__

.github/workflows/pr-validation.yml

@@ -31,9 +31,7 @@ jobs:
         pip install -r requirements.txt
     - name: Lint with flake8
       run: |
-        # flake8 settings are in setup.cfg
         flake8 . --count --show-source --statistics --exit-zero
-    - name: Test with pytest
+    - name: Pytest unit tests
       run: |
-        # pytest markers are defined in setup.cfg
         pytest -m "not e2e" --verbose

CHANGELOG.md

@@ -0,0 +1,8 @@
+## v0.1.0
+
+Initial release, which includes the following features:
+
+- Orchestrations and activities
+- Durable timers
+- Sub-orchestrations
+

Makefile

@@ -0,0 +1,17 @@
+init:
+	pip3 install -r requirements.txt
+
+test-unit:
+	pytest -m "not e2e" --verbose
+
+test-e2e:
+	pytest -m e2e --verbose
+
+install:
+	python3 -m pip install .
+
+gen-proto:
+# NOTE: There is currently a hand-edit that we make to the generated orchestrator_service_pb2.py file after it's generated to help resolve import problems.
+	python3 -m grpc_tools.protoc --proto_path=./submodules/durabletask-protobuf/protos  --python_out=./durabletask/internal --pyi_out=./durabletask/internal --grpc_python_out=./durabletask/internal orchestrator_service.proto
+
+.PHONY: init test-unit test-e2e gen-proto install

README.md

@@ -9,46 +9,131 @@ This repo contains a Python client SDK for use with the [Durable Task Framework
 
 > Note that this project is **not** currently affiliated with the [Durable Functions](https://docs.microsoft.com/azure/azure-functions/durable/durable-functions-overview) project for Azure Functions. If you are looking for a Python SDK for Durable Functions, please see [this repo](https://github.com/Azure/azure-functions-durable-python).
 
+## Features
+
+- [x] Orchestrations
+- [x] Activities
+- [x] Durable timers
+- [x] Sub-orchestrations
+- [ ] External events
+- [ ] Suspend and resume
+- [ ] Retry policies
+
+## Supported patterns
+
+The following orchestration patterns are currently supported.
+
+### Function chaining
+
+An orchestration can chain a sequence of function calls using the following syntax:
+
+```python
+# simple activity function that returns a greeting
+def hello(ctx: task.ActivityContext, name: str) -> str:
+    return f'Hello {name}!'
+
+# orchestrator function that sequences the activity calls
+def sequence(ctx: task.OrchestrationContext, _):
+    result1 = yield ctx.call_activity(hello, input='Tokyo')
+    result2 = yield ctx.call_activity(hello, input='Seattle')
+    result3 = yield ctx.call_activity(hello, input='London')
+
+    return [result1, result2, result3]
+```
+
+You can find the full sample [here](./examples/activity_sequence.py).
+
+### Fan-out/fan-in
+
+An orchestration can fan-out a dynamic number of function calls in parallel and then fan-in the results using the following syntax:
+
+```python
+# activity function for getting the list of work items
+def get_work_items(ctx: task.ActivityContext, _) -> List[str]:
+    # ...
+
+# activity function for processing a single work item
+def process_work_item(ctx: task.ActivityContext, item: str) -> int:
+    # ...
+
+# orchestrator function that fans-out the work items and then fans-in the results
+def orchestrator(ctx: task.OrchestrationContext, _):
+    # the number of work-items is unknown in advance
+    work_items = yield ctx.call_activity(get_work_items)
+
+    # fan-out: schedule the work items in parallel and wait for all of them to complete
+    tasks = [ctx.call_activity(process_work_item, input=item) for item in work_items]
+    results = yield task.when_all(tasks)
+
+    # fan-in: summarize and return the results
+    return {'work_items': work_items, 'results': results, 'total': sum(results)}
+```
+
+You can find the full sample [here](./examples/fanout_fanin.py).
+
 ## Getting Started
 
 ### Prerequisites
 
 - Python 3.10 or higher
+- A Durable Task-compatible sidecar, like [Dapr Workflow](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-overview/)
+
+### Installing the Durable Task Python client SDK
+
+Installation is currently only supported from source. Ensure pip, setuptools, and wheel are up-to-date.
+
+```sh
+python3 -m pip install --upgrade pip setuptools wheel
+```
 
-### Installing
+To install this package from source, clone this repository and run the following command from the project root:
 
-#### Install from PyPI
-This package is not yet published to [PyPI](https://pypi.org/).
+```sh
+python3 -m pip install .
+```
 
-#### Install from source
-TODO
+### Run the samples
+
+See the [examples](./examples) directory for a list of sample orchestrations and instructions on how to run them.
 
 ## Development
-The following is more information about how to develop this project.
+
+The following is more information about how to develop this project. Note that development commands require that `make` is installed on your local machine. If you're using Windows, you can install `make` using [Chocolatey](https://chocolatey.org/) or use WSL.
 
 ### Generating protobufs
-If the gRPC proto definitions need to be updated, the corresponding source code can be regenerated using the following command from the project root:
 
-```bash
-python3 -m grpc_tools.protoc --proto_path=./submodules/durabletask-protobuf/protos  --python_out=./durabletask/protos --pyi_out=./durabletask/protos --grpc_python_out=./durabletask/protos orchestrator_service.proto
+Protobuf definitions are stored in the [./submodules/durabletask-proto](./submodules/durabletask-proto) directory, which is a submodule. To update the submodule, run the following command from the project root:
+
+```sh
+git submodule update --init
 ```
 
-### Linting and running unit tests
+Once the submodule is available, the corresponding source code can be regenerated using the following command from the project root:
+
+```sh
+make proto-gen
+```
 
-See the [pr-validation.yml](.github/workflows/pr-validation.yml) workflow for the full list of commands that are run as part of the CI/CD pipeline.
+### Running unit tests
+
+Unit tests can be run using the following command from the project root. Unit tests _don't_ require a sidecar process to be running.
+
+```sh
+make test-unit
+```
 
 ### Running E2E tests
 
-The E2E (end-to-end) tests require a sidecar process to be running. You can run a sidecar process using the following `docker` command (assumes you have Docker installed on your local system.)
+The E2E (end-to-end) tests require a sidecar process to be running. You can use the Dapr sidecar for this or run a Durable Task test sidecar using the following `docker` command:
 
 ```sh
 docker run --name durabletask-sidecar -p 4001:4001 --env 'DURABLETASK_SIDECAR_LOGLEVEL=Debug' --rm cgillum/durabletask-sidecar:latest start --backend Emulator
 ```
 
-To run the E2E tests, run the following command from the project root (we assume you have `pytest` installed locally):
+To run the E2E tests, run the following command from the project root:
 
 ```sh
-pytest -m e2e --verbose
+make test-e2e
 ```
 
 ## Contributing

durabletask/__init__.py

@@ -3,4 +3,5 @@
 
 """Durable Task SDK for Python"""
 
+
 PACKAGE_NAME = "durabletask"

durabletask/api/state.py

@@ -3,50 +3,93 @@
 
 import logging
 import uuid
+from dataclasses import dataclass
 from datetime import datetime
+from enum import Enum
 from typing import TypeVar
 
 import grpc
 import simplejson as json
 from google.protobuf import wrappers_pb2
 
+import durabletask.internal.helpers as helpers
+import durabletask.internal.orchestrator_service_pb2 as pb
+import durabletask.internal.orchestrator_service_pb2_grpc as stubs
 import durabletask.internal.shared as shared
-import durabletask.protos.helpers as helpers
-import durabletask.protos.orchestrator_service_pb2 as pb
-import durabletask.task.registry as registry
-from durabletask.api.state import OrchestrationState, new_orchestration_state
-from durabletask.protos.orchestrator_service_pb2_grpc import \
-    TaskHubSidecarServiceStub
-from durabletask.task.orchestration import Orchestrator
+from durabletask import task
 
 TInput = TypeVar('TInput')
 TOutput = TypeVar('TOutput')
 
 
+class OrchestrationStatus(Enum):
+    """The status of an orchestration instance."""
+    RUNNING = pb.ORCHESTRATION_STATUS_RUNNING
+    COMPLETED = pb.ORCHESTRATION_STATUS_COMPLETED
+    FAILED = pb.ORCHESTRATION_STATUS_FAILED
+    TERMINATED = pb.ORCHESTRATION_STATUS_TERMINATED
+    CONTINUED_AS_NEW = pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
+    PENDING = pb.ORCHESTRATION_STATUS_PENDING
+    SUSPENDED = pb.ORCHESTRATION_STATUS_SUSPENDED
+
+    def __str__(self):
+        return helpers.get_orchestration_status_str(self.value)
+
+
+@dataclass
+class OrchestrationState:
+    instance_id: str
+    name: str
+    runtime_status: OrchestrationStatus
+    created_at: datetime
+    last_updated_at: datetime
+    serialized_input: str | None
+    serialized_output: str | None
+    serialized_custom_status: str | None
+    failure_details: pb.TaskFailureDetails | None
+
+
+def new_orchestration_state(instance_id: str, res: pb.GetInstanceResponse) -> OrchestrationState | None:
+    if not res.exists:
+        return None
+
+    state = res.orchestrationState
+    return OrchestrationState(
+        instance_id,
+        state.name,
+        OrchestrationStatus(state.orchestrationStatus),
+        state.createdTimestamp.ToDatetime(),
+        state.lastUpdatedTimestamp.ToDatetime(),
+        state.input.value if not helpers.is_empty(state.input) else None,
+        state.output.value if not helpers.is_empty(state.output) else None,
+        state.customStatus.value if not helpers.is_empty(state.customStatus) else None,
+        state.failureDetails if state.failureDetails.errorMessage != '' or state.failureDetails.errorType != '' else None)
+
+
 class TaskHubGrpcClient:
 
     def __init__(self, *,
                  host_address: str | None = None,
                  log_handler=None,
                  log_formatter: logging.Formatter | None = None):
         channel = shared.get_grpc_channel(host_address)
-        self._stub = TaskHubSidecarServiceStub(channel)
+        self._stub = stubs.TaskHubSidecarServiceStub(channel)
         self._logger = shared.get_logger(log_handler, log_formatter)
 
-    def schedule_new_orchestration(self, orchestrator: Orchestrator[TInput, TOutput], *,
+    def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOutput], *,
                                    input: TInput | None = None,
                                    instance_id: str | None = None,
                                    start_at: datetime | None = None) -> str:
 
-        name = registry.get_name(orchestrator)
+        name = task.get_name(orchestrator)
 
         req = pb.CreateInstanceRequest(
             name=name,
             instanceId=instance_id if instance_id else uuid.uuid4().hex,
             input=wrappers_pb2.StringValue(value=json.dumps(input)) if input else None,
             scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None)
 
-        self._logger.info(f"Starting new '{name}' instance with ID = '{instance_id}'.")
+        self._logger.info(f"Starting new '{name}' instance with ID = '{req.instanceId}'.")
         res: pb.CreateInstanceResponse = self._stub.StartInstance(req)
         return res.instanceId
 
@@ -60,7 +103,7 @@ def wait_for_orchestration_start(self, instance_id: str, *,
                                      timeout: int = 60) -> OrchestrationState | None:
         req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
         try:
-            self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to start.")
+            self._logger.info(f"Waiting up to {timeout}s for instance '{instance_id}' to start.")
             res: pb.GetInstanceResponse = self._stub.WaitForInstanceStart(req, timeout=timeout)
             return new_orchestration_state(req.instanceId, res)
         except grpc.RpcError as rpc_error:

durabletask/api/task_hub_client.py durabletask/client.pydurabletask/api/task_hub_client.py renamed to durabletask/client.py

@@ -8,7 +8,7 @@
 import simplejson as json
 from google.protobuf import timestamp_pb2, wrappers_pb2
 
-import durabletask.protos.orchestrator_service_pb2 as pb
+import durabletask.internal.orchestrator_service_pb2 as pb
 
 # TODO: The new_xxx_event methods are only used by test code and should be moved elsewhere
 
@@ -175,3 +175,12 @@ def new_create_sub_orchestration_action(
 
 def is_empty(v: wrappers_pb2.StringValue):
     return v is None or v.value == ''
+
+
+def get_orchestration_status_str(status: pb.OrchestrationStatus):
+    try:
+        const_name = pb.OrchestrationStatus.Name(status)
+        if const_name.startswith('ORCHESTRATION_STATUS_'):
+            return const_name[len('ORCHESTRATION_STATUS_'):]
+    except Exception:
+        return "UNKNOWN"