activities.py

import asyncio
from dataclasses import dataclass

from temporalio import activity


@dataclass
class IncompatibleActivityInput:
    """Input for the incompatible activity."""

    called_by: str
    more_data: str


@activity.defn
async def some_activity(called_by: str) -> str:
    """Basic activity for the workflow."""
    return f"some_activity called by {called_by}"


@activity.defn
async def some_incompatible_activity(input_data: IncompatibleActivityInput) -> str:
    """Incompatible activity that takes different input."""
    return f"some_incompatible_activity called by {input_data.called_by} with {input_data.more_data}"


@activity.defn
async def slow_activity(duration_seconds: int = 5) -> str:
    """Slow activity for load testing - simulates work that takes time.
    
    This activity sleeps for the specified duration, which helps generate
    backlog when many workflows call it simultaneously.
    """
    activity.logger.info(f"Starting slow activity, will take {duration_seconds}s")
    await asyncio.sleep(duration_seconds)
    activity.logger.info(f"Slow activity completed after {duration_seconds}s")
    return f"slow_activity completed after {duration_seconds} seconds"