This sample is a Python implementation of the [TypeScript "Sticky Workers" example](https://github.com/temporalio/samples-typescript/tree/main/activities-sticky-queues), full credit for the design to the authors of that sample. A [sticky execution](https://docs.temporal.io/tasks#sticky-execution) is a job distribution design pattern where all workflow computational tasks are executed on a single worker. In the Go and Java SDKs this is explicitly supported via the Session option, but in other SDKs a different approach is required.

Typical use cases for sticky executions include tasks where interaction with a filesystem is required, such as data processing or interacting with legacy access structures. This example will write text files to folders corresponding to each worker, located in the `demo_fs` folder. In production, these folders would typically be independent machines in a worker cluster.

This strategy is:

- Create a `get_available_task_queue` activity that generates a unique task queue name, `unique_worker_task_queue`.

- For activities intended to be "sticky", only register them in one Worker, and have that be the only Worker listening on that `unique_worker_task_queue`. This will be run on a series of `FileProcessing` workflows.

- Execute workflows from the Client like normal. Check the Temporal Web UI to confirm tasks were staying with their respective worker.

It doesn't matter where the `get_available_task_queue` activity is run, so it can be "non sticky" as per Temporal default behavior. In this demo, `unique_worker_task_queue` is simply a `uuid` initialized in the Worker, but you can inject smart logic here to uniquely identify the Worker, [as Netflix did](https://community.temporal.io/t/using-dynamic-task-queues-for-traffic-routing/3045). Our example differs from the Node sample by running across 5 unique task queues.

Activities have been artificially slowed with `time.sleep(3)` to simulate slow activities.

To run, first see [README.md](../README.md) for prerequisites. Then, run the following from this directory to start the

worker:

poetry run python worker.py

This will start the worker. Then, in another terminal, run the following to execute the workflow:

poetry run python starter.py

(temporalio-samples-py3.10) user@machine:~/samples-python/activities_sticky_queues$ poetry run python starter.py

Output checksums:

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

49d7419e6cba3575b3158f62d053f922aa08b23c64f05411cda3213b56c84ba4

<details>

<summary>Checking the history to see where activities are run</summary>

All activities for the one workflow are running against the same task queue, which corresponds to unique workers:


</details>

import asyncio

from uuid import uuid4

from temporalio.client import Client

from activity_sticky_queues.tasks import FileProcessing

async def main():

# Connect client

client = await Client.connect("localhost:7233")

# Start 10 concurrent workflows

futures = []

for idx in range(10):

result = client.execute_workflow(

FileProcessing.run,

id=f"activity_sticky_queue-workflow-id-{idx}",

task_queue="activity_sticky_queue-distribution-queue",

)

await asyncio.sleep(0.1)

futures.append(result)

checksums = await asyncio.gather(*futures)

print("\n".join([f"Output checksums:"] + checksums))

if __name__ == "__main__":

asyncio.run(main())

import asyncio

from dataclasses import dataclass

from datetime import timedelta

from hashlib import sha256

from pathlib import Path

from temporalio import activity, workflow

from temporalio.common import RetryPolicy

def _get_delay_secs() -> float:

return 3

def _get_local_path() -> Path:

return Path(__file__).parent / "demo_fs"

def write_file(path: Path, body: str) -> None:

"""Convenience write wrapper for mocking FS"""

with open(path, "w") as handle:

handle.write(body)

def read_file(path) -> bytes:

"""Convenience read wrapper for mocking FS"""

with open(path, "rb") as handle:

return handle.read()

def delete_file(path) -> None:

"""Convenience delete wrapper for mocking FS"""

Path(path).unlink()

def create_filepath(unique_worker_id: str, workflow_uuid: str) -> Path:

"""Creates required folders and builds filepath"""

directory = _get_local_path() / unique_worker_id

directory.mkdir(parents=True, exist_ok=True)

filepath = directory / workflow_uuid

return filepath

def process_file_contents(file_content: bytes) -> str:

"""Returns hash of file string"""

return sha256(file_content).hexdigest()

class DownloadObj:

url: str

unique_worker_id: str

workflow_uuid: str

async def get_available_task_queue() -> str:

"""Just a stub for typedworkflow invocation."""

raise NotImplementedError

async def download_file_to_worker_filesystem(details: DownloadObj) -> str:

"""Simulates downloading a file to a local filesystem"""

# FS ops

path = create_filepath(details.unique_worker_id, details.workflow_uuid)

activity.logger.info(f"Downloading ${details.url} and saving to ${path}")

# Here is where the real download code goes. Developers should be careful

# not to block an async activity. If there are concerns about blocking download

# or disk IO, developers should use loop.run_in_executor or change this activity

# to be synchronous. Also like for all non-immediate activities, be sure to

# heartbeat during download.

await asyncio.sleep(_get_delay_secs())

body = "downloaded body"

write_file(path, body)

return str(path)

async def work_on_file_in_worker_filesystem(path: str) -> str:

"""Processing the file, in this case identical MD5 hashes"""

content = read_file(path)

checksum = process_file_contents(content)

await asyncio.sleep(_get_delay_secs())

activity.logger.info(f"Did some work on {path}, checksum {checksum}")

return checksum

async def clean_up_file_from_worker_filesystem(path: str) -> None:

"""Deletes the file created in the first activity, but leaves the folder"""

await asyncio.sleep(_get_delay_secs())

activity.logger.info(f"Removing {path}")

delete_file(path)

class FileProcessing:

@workflow.run

async def run(self) -> str:

"""Workflow implementing the basic file processing example.

workflow.logger.info("Searching for available worker")

unique_worker_task_queue = await workflow.execute_activity(

activity=get_available_task_queue,

start_to_close_timeout=timedelta(seconds=10),

)

workflow.logger.info(f"Matching workflow to worker {unique_worker_task_queue}")

download_params = DownloadObj(

url="http://temporal.io",

unique_worker_id=unique_worker_task_queue,

workflow_uuid=str(workflow.uuid4()),

)

download_path = await workflow.execute_activity(

download_file_to_worker_filesystem,

download_params,

start_to_close_timeout=timedelta(seconds=10),

task_queue=unique_worker_task_queue,

)

checksum = "failed execution" # Sentinel value

try:

checksum = await workflow.execute_activity(

work_on_file_in_worker_filesystem,

download_path,

start_to_close_timeout=timedelta(seconds=10),

retry_policy=RetryPolicy(

maximum_attempts=1,

# maximum_interval=timedelta(milliseconds=500),

),

task_queue=unique_worker_task_queue,

)

finally:

await workflow.execute_activity(

clean_up_file_from_worker_filesystem,

download_path,

start_to_close_timeout=timedelta(seconds=10),

task_queue=unique_worker_task_queue,

)

return checksum

import asyncio

import logging # noqa

import random

from typing import List

from uuid import UUID

from temporalio import activity

from temporalio.client import Client

from temporalio.worker import Worker

from activity_sticky_queues import tasks

interrupt_event = asyncio.Event()

async def main():

# Uncomment the line below to see logging

# logging.basicConfig(level=logging.INFO)

# Comment line to see non-deterministic functionality

random.seed(667)

# Create random task queues and build task queue selection function

task_queues: List[str] = [

f"activity_sticky_queue-host-{UUID(int=random.getrandbits(128))}"

for _ in range(5)

]

@activity.defn(name="get_available_task_queue")

async def select_task_queue_random() -> str:

"""Randomly assign the job to a queue"""

return random.choice(task_queues)

# Start client

client = await Client.connect("localhost:7233")

# Run a worker to distribute the workflows

run_futures = []

handle = Worker(

client,

task_queue="activity_sticky_queue-distribution-queue",

workflows=[tasks.FileProcessing],

activities=[select_task_queue_random],

)

run_futures.append(handle.run())

print("Base worker started")

# Run the workers for the individual task queues

for queue_id in task_queues:

handle = Worker(

client,

task_queue=queue_id,

activities=[

tasks.download_file_to_worker_filesystem,

tasks.work_on_file_in_worker_filesystem,

tasks.clean_up_file_from_worker_filesystem,

],

)

run_futures.append(handle.run())

# Wait until interrupted

print(f"Worker {queue_id} started")

print("All workers started, ctrl+c to exit")

await asyncio.gather(*run_futures)

if __name__ == "__main__":

loop = asyncio.new_event_loop()

try:

loop.run_until_complete(main())

except KeyboardInterrupt:

interrupt_event.set()

loop.run_until_complete(loop.shutdown_asyncgens())

print("\nShutting down workers")

from pathlib import Path

from unittest import mock

from activity_sticky_queues import tasks

RETURNED_PATH = "valid/path"

tasks._get_delay_secs = mock.MagicMock(return_value=0.0001)

tasks._get_local_path = mock.MagicMock(return_value=Path(RETURNED_PATH))

async def test_download_activity():

worker_id = "an-id"

workflow_uuid = "uuid"

want = Path(RETURNED_PATH) / worker_id / workflow_uuid

details = tasks.DownloadObj("tdd.com", worker_id, workflow_uuid)

with mock.patch.object(tasks, "write_file") as mock_write:

response = await tasks.download_file_to_worker_filesystem(details)

assert Path(response) == want

mock_write.assert_called_once()

async def test_processing_activity():

file_contents = b"contents"

want = tasks.process_file_contents(file_contents)

with mock.patch.object(tasks, "read_file", return_value=file_contents):

response = await tasks.work_on_file_in_worker_filesystem(RETURNED_PATH)

assert response == want

async def test_clean_up_activity():

with mock.patch.object(tasks, "delete_file") as mock_delete:

await tasks.clean_up_file_from_worker_filesystem(RETURNED_PATH)

mock_delete.assert_called_once_with(RETURNED_PATH)