forze-temporal-workflows

Forze Temporal workflows

Use when starting, signaling, updating, querying, or testing workflow-backed handlers. The core application depends on durable workflow contracts; forze_temporal supplies the Temporal adapter.

Workflow spec

from enum import StrEnum

from forze.application.contracts.durable.workflow import (
    DurableWorkflowInvokeSpec,
    DurableWorkflowSignalSpec,
    DurableWorkflowSpec,
)


class WorkflowName(StrEnum):
    PROJECT_ONBOARDING = "project-onboarding"

project_onboarding = DurableWorkflowSpec( name=WorkflowName.PROJECT_ONBOARDING, run=DurableWorkflowInvokeSpec(args_type=StartOnboarding, return_type=OnboardingResult), signals={ "step_completed": DurableWorkflowSignalSpec( name="step_completed", args_type=StepCompleted, ) }, )


The spec name is the dependency route and should match `TemporalDepsModule.workflows`.

## Runtime wiring

```python
from forze.application.execution import DepsRegistry, LifecyclePlan
from forze_temporal import (
    TemporalClient,
    TemporalConfig,
    TemporalDepsModule,
    temporal_lifecycle_step,
)

temporal_client = TemporalClient()
temporal_module = TemporalDepsModule(
    client=temporal_client,
    workflows={
        WorkflowName.PROJECT_ONBOARDING: {
            "queue": "project-tasks",
            "tenant_aware": True,
        }
    },
)

deps = DepsRegistry.from_modules(temporal_module)
lifecycle = LifecyclePlan.from_steps(
    temporal_lifecycle_step(
        host="localhost:7233",
        config=TemporalConfig(namespace="default"),
    )
)

If workflows is empty, only TemporalClientDepKey is registered.

Handler resolution

There is no ctx.workflow_command(...) helper. Resolve workflow ports with ctx.deps.resolve_configurable.

import attrs

from forze.application.contracts.durable.workflow import (
    DurableWorkflowCommandDepKey,
    DurableWorkflowCommandPort,
    DurableWorkflowHandle,
    DurableWorkflowQueryDepKey,
    DurableWorkflowQueryPort,
    DurableWorkflowRunDescription,
)
from forze.application.contracts.execution import Handler


@attrs.define(slots=True, kw_only=True, frozen=True)
class StartProjectOnboarding(Handler[StartOnboarding, DurableWorkflowHandle]):
    commands: DurableWorkflowCommandPort

    async def __call__(self, args: StartOnboarding) -> DurableWorkflowHandle:
        return await self.commands.start(
            args,
            workflow_id=f"project:{args.project_id}",
        )


@attrs.define(slots=True, kw_only=True, frozen=True)
class GetOnboardingStatus(Handler[DurableWorkflowHandle, DurableWorkflowRunDescription]):
    queries: DurableWorkflowQueryPort

    async def __call__(self, args: DurableWorkflowHandle) -> DurableWorkflowRunDescription:
        return await self.queries.describe(args)


@attrs.define(slots=True, kw_only=True, frozen=True)
class GetOnboardingResult(Handler[DurableWorkflowHandle, OnboardingResult]):
    queries: DurableWorkflowQueryPort

    async def __call__(self, args: DurableWorkflowHandle) -> OnboardingResult:
        return await self.queries.result(args)


# Register on OperationRegistry:
# lambda ctx: StartProjectOnboarding(
#     commands=ctx.deps.resolve_configurable(
#         ctx, DurableWorkflowCommandDepKey, project_onboarding, route=project_onboarding.name
#     ),
# )

Context and tenancy

Use ExecutionContextInterceptor and TemporalContextCodec when InvocationMetadata, AuthnIdentity, or TenantIdentity must cross Temporal client/worker boundaries. For tenant-aware workflow config, bind TenantIdentity before starting workflows so generated IDs can include tenant scope.

Workflow schedules (Temporal Schedules API)

Resolve schedule ports the same way as workflow command/query ports:

from datetime import timedelta

from forze.application.contracts.durable.workflow import (
    DurableWorkflowScheduleCommandDepKey,
    DurableWorkflowScheduleTiming,
)

schedules = ctx.deps.resolve_configurable(
    ctx,
    DurableWorkflowScheduleCommandDepKey,
    project_onboarding,
    route=project_onboarding.name,
)
await schedules.upsert(
    "nightly-onboarding",
    StartOnboarding(project_id="p1"),
    DurableWorkflowScheduleTiming(cron_expressions=("0 2 * * *",)),
)

Declarative bootstrap on deploy:

from forze.application.contracts.durable.workflow import DurableWorkflowScheduleBootstrap

TemporalDepsModule(
    client=client,
    workflows={WorkflowName.PROJECT_ONBOARDING: {"queue": "project-tasks"}},
    schedule_bootstraps=[
        DurableWorkflowScheduleBootstrap(
            workflow_name=WorkflowName.PROJECT_ONBOARDING,
            schedule_id="nightly",
            default_args=StartOnboarding(project_id="default"),
            timing=DurableWorkflowScheduleTiming(interval=timedelta(hours=24)),
        ),
    ],
)

# Pass the same workflow config map to lifecycle:
temporal_lifecycle_step(host="...", workflow_configs={...})

Schedules require a Temporal server with the Schedules API (not the time-skipping test environment).

Testing

For handlers that only need to verify a workflow command was issued, register fake DurableWorkflowCommandDepKey / DurableWorkflowQueryDepKey factories in Deps (for example via MockDepsModule or a test DepsRegistry). Schedule handlers can use fake DurableWorkflowScheduleCommandDepKey / DurableWorkflowScheduleQueryDepKey factories similarly. For workflow definition tests, follow the Temporal integration testing section.

Anti-patterns

Looking for ctx.workflow_* helpers — use ctx.deps.resolve_configurable(ctx, DurableWorkflowCommandDepKey, spec, route=spec.name).
Putting Temporal task queues in DurableWorkflowSpec — queues belong in TemporalWorkflowConfig.
Using raw Temporal SDK types in handlers — keep handlers on DurableWorkflowSpec, ports, and Pydantic DTOs.
Skipping lifecycle — the client connects through temporal_lifecycle_step.
Ignoring context propagation in workers — register interceptors on both client and worker when identity/correlation matters.

About this skill