conductor-python 1.1.4

Netflix Conductor Python SDK

Conductor OSS Python SDK

Python SDK for working with https://github.com/conductor-oss/conductor

Conductor is an open source distributed, scalable and highly available orchestration platform that allows developers to build powerful distributed applications. You can find the documentation for Conductor here: Conductor Docs

⭐ Conductor OSS

Show support for the Conductor OSS. Please help spread the awareness by starring Conductor repo.

Content

• Install SDK
  • Get Conductor Python SDK
  • Setup SDK
• Start Conductor Server
• Simple Hello World Application using Conductor
  • Step 1: Create a Workflow
  • Step 2: Write Worker
  • Step 3: Write your application
• Using Conductor in your application
  • Create and Run Conductor Workers
  • Create Conductor Workflows
  • Using Conductor in your Application

Install SDK

Create a virtual environment to build your package

virtualenv conductor
source conductor/bin/activate

Get Conductor Python SDK

SDK needs Python 3.9+.

python3 -m pip install conductor-python

Setup SDK

Point the SDK to the Conductor Server API endpoint

export CONDUCTOR_SERVER_URL=http://localhost:8080/api

(Optionally) If you are using a Conductor server that requires authentication

How to obtain the key and secret from the conductor server

export CONDUCTOR_AUTH_KEY=your_key
export CONDUCTOR_AUTH_SECRET=your_key_secret

Start Conductor Server

docker run --init -p 8080:8080 -p 5000:5000 conductoross/conductor-standalone:3.15.0

After starting the server navigate to http://localhost:1234 to ensure the server has started successfully.

Simple Hello World Application using Conductor

In this section, we will create a simple "Hello World" application that uses Conductor.

Step 1: Create a Workflow

Use Code to create workflows

Create greetings_workflow.py with the following:

from conductor.client.workflow.conductor_workflow import ConductorWorkflow
from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
from greetings import greet

def greetings_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    name = 'hello'
    workflow = ConductorWorkflow(name=name, executor=workflow_executor)
    workflow.version = 1
    workflow >> greet(task_ref_name='greet_ref', name=workflow.input('name'))
    return workflow

(alternatively) Use JSON to create workflows

Create workflow.json with the following:

{
  "name": "hello",
  "description": "hello workflow",
  "version": 1,
  "tasks": [
    {
      "name": "greet",
      "taskReferenceName": "greet_ref",
      "type": "SIMPLE",
      "inputParameters": {
        "name": "${workflow.input.name}"
      }
    }
  ],
  "timeoutPolicy": "TIME_OUT_WF",
  "timeoutSeconds": 60
}

Now, register this workflow with the server:

curl -X POST -H "Content-Type:application/json" \
http://localhost:8080/api/metadata/workflow -d @workflow.json

Step 2: Write Worker

Create greetings.py with a simple worker and a workflow function.

[!note] A single workflow application can have workers written in different languages.

from conductor.client.worker.worker_task import worker_task


@worker_task(task_definition_name='greet')
def greet(name: str) -> str:
    return f'Hello my friend {name}'

Step 3: Write your application

Let's add greetings_main.py with the main method:

from conductor.client.automator.task_handler import TaskHandler
from conductor.client.configuration.configuration import Configuration
from conductor.client.workflow.conductor_workflow import ConductorWorkflow
from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor
from greetings_workflow import greetings_workflow


def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    workflow = greetings_workflow(workflow_executor=workflow_executor)
    workflow.register(True)
    return workflow


def main():
    # points to http://localhost:8080/api by default
    api_config = Configuration()

    workflow_executor = WorkflowExecutor(configuration=api_config)

    # Needs to be done only when registering a workflow one-time
    workflow = register_workflow(workflow_executor)

    task_handler = TaskHandler(configuration=api_config)
    task_handler.start_processes()

    workflow_run = workflow_executor.execute(name=workflow.name, version=workflow.version,
                                             workflow_input={'name': 'Orkes'})

    print(f'\nworkflow result: {workflow_run.output["result"]}\n')
    print(f'see the workflow execution here: {api_config.ui_host}/execution/{workflow_run.workflow_id}\n')
    task_handler.stop_processes()


if __name__ == '__main__':
    main()

[!NOTE] That's it - you just created your first distributed python app!

Using Conductor in your application

There are three main ways you will use Conductor when building durable, resilient, distributed applications.

1. Write service workers that implements business logic to accomplish a specific goal - such as initiate payment transfer, get user information from database etc.
2. Create Conductor workflows that implements application state - A typical workflow implements SAGA pattern
3. Use Conductor SDK and APIs to manage workflows from your application.

In this guide, we will dive deeper into each of these topic.

Create and Run Conductor Workers

Create Conductor Workflows

Using Conductor in your Application