Simple Python Gists

define_workflow.py

@workflow.defn
class SayHello:
    @workflow.run
    async def run(self, name: str) -> str:
        return f"Hello, {name}!"

define_workflow_and_activity.py

@activity.defn
async def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@workflow.defn
class SayHello:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            say_hello, name, schedule_to_close_timeout=timedelta(seconds=5)
        )

run_workflow.py

async def run_workflow():
    client = await Client.connect("localhost:7233")
    result = await client.execute_workflow(
        SayHello.run,
        "Temporal",
        id="my-workflow-id",
        task_queue="my-task-queue".
    )
    print(f"Result: {result}")