Getting Started with Durable Workflows

The nala.athomic.workflows module provides an agnostic way to orchestrate long-running business processes. It allows you to write workflow logic once and execute it either on a production-grade engine (Temporal.io) or in-memory for lightning-fast tests (Local).

1. Defining a Workflow

Workflows are defined using the standard Temporal Python SDK decorators. This ensures compatibility with the production engine.

from datetime import timedelta
from temporalio import activity, workflow

# 1. Define Activities (The "Side Effects")
# Activities are where you perform I/O (DB calls, HTTP requests).
@activity.defn
async def charge_user(user_id: str, amount: float) -> str:
    # Logic to call payment gateway...
    return "transaction_id_123"

@activity.defn
async def send_receipt(email: str) -> None:
    # Logic to send email...
    pass

# 2. Define the Workflow (The "Orchestrator")
# Workflows must be deterministic (no random, no system time, no direct I/O).
@workflow.defn(name="PaymentProcessing")
class PaymentWorkflow:
    @workflow.run
    async def run(self, user_id: str, email: str, amount: float) -> dict:
        workflow.logger.info(f"Starting payment for {user_id}")

        # Execute Activity 1: Charge User
        # Automatic retries are handled by the engine here.
        tx_id = await workflow.execute_activity(
            charge_user,
            args=[user_id, amount],
            start_to_close_timeout=timedelta(seconds=10),
        )

        # Execute Activity 2: Send Receipt
        await workflow.execute_activity(
            send_receipt,
            email,
            start_to_close_timeout=timedelta(seconds=5),
        )

        return {"status": "COMPLETED", "transaction_id": tx_id}

---

2. Registering for Local Execution

When running in Production (Temporal), the Worker automatically scans and registers workflows. However, when running in Local/Test Mode (In-Memory), the LocalWorkflowClient does not scan code. You must explicitly register your workflow function in the in-memory registry.

This is typically done in your conftest.py or application startup logic for dev environments.

from nala.athomic.workflows.providers.local.registry import local_workflow_registry
from my_app.workflows.payment import PaymentWorkflow

# Register the workflow name pointing to the class method
# This allows client.start_workflow("PaymentProcessing", ...) to work locally.
local_workflow_registry.register("PaymentProcessing", PaymentWorkflow().run)

---

3. Executing a Workflow

To start a workflow, use the agnostic WorkflowClientFactory. This keeps your business logic decoupled from the specific backend (Temporal vs. Local) configured in your settings.toml.

import asyncio
from nala.athomic.workflows.factory import WorkflowClientFactory
from nala.athomic.workflows.models import WorkflowOptions

async def trigger_process():
    # 1. Create the client (Backend is determined by settings.WORKFLOWS.PROVIDER.BACKEND)
    client = WorkflowClientFactory.create()

    # 2. Configure Execution Options
    # Note: 'execution_timeout_seconds' accepts strings like "1h", "30m"
    options = WorkflowOptions(
        workflow_id=f"pay-req-{uuid.uuid4()}",
        task_queue="payments-queue",
        execution_timeout_seconds="1h" 
    )

    # 3. Start Execution
    # The payload (arguments) is automatically serialized, compressed, 
    # and encrypted by the Nala Payload Pipeline before leaving the app.
    execution = await client.start_workflow(
        "PaymentProcessing", # Matches the name in @workflow.defn
        "user_123",          # Arg 1
        "user@example.com",  # Arg 2
        99.90,               # Arg 3
        options=options
    )

    print(f"Workflow started with RunID: {execution.run_id}")

    # 4. (Optional) Wait for result
    result = await client.get_result(execution.workflow_id, run_id=execution.run_id)
    print(f"Result: {result.result}")