workerv2.py

"""Worker v2 for the worker versioning sample."""

import asyncio
import logging
import os

from temporalio.client import Client
from temporalio.common import WorkerDeploymentVersion
from temporalio.worker import Worker, WorkerDeploymentConfig

from worker_versioning.activities import some_activity, some_incompatible_activity
from worker_versioning.workflows import AutoUpgradingWorkflowV1b, PinnedWorkflowV2

logging.basicConfig(level=logging.INFO)

# Configuration from environment (for Kubernetes deployment)
TEMPORAL_ADDRESS = os.getenv("TEMPORAL_ADDRESS", "localhost:7233")
TEMPORAL_NAMESPACE = os.getenv("TEMPORAL_NAMESPACE", "default")
TASK_QUEUE = os.getenv("TEMPORAL_TASK_QUEUE", "worker-versioning")
DEPLOYMENT_NAME = os.getenv("TEMPORAL_DEPLOYMENT_NAME", "my-deployment")

# Build ID for this worker version
BUILD_ID = "2.0"


async def main() -> None:
    """Run worker v2."""
    logging.info(f"Connecting to Temporal at {TEMPORAL_ADDRESS}")
    logging.info(f"Namespace: {TEMPORAL_NAMESPACE}, Task Queue: {TASK_QUEUE}")
    
    client = await Client.connect(TEMPORAL_ADDRESS, namespace=TEMPORAL_NAMESPACE)

    # Create worker v2
    worker = Worker(
        client,
        task_queue=TASK_QUEUE,
        workflows=[AutoUpgradingWorkflowV1b, PinnedWorkflowV2],
        activities=[some_activity, some_incompatible_activity],
        deployment_config=WorkerDeploymentConfig(
            version=WorkerDeploymentVersion(
                deployment_name=DEPLOYMENT_NAME, build_id=BUILD_ID
            ),
            use_worker_versioning=True,
        ),
    )

    logging.info(f"Starting worker v2 (build {BUILD_ID}) with deployment {DEPLOYMENT_NAME}")
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())