workflows.services.context

Context assembly utilities for workflow templates and executors.

This module builds the per-instance template context (ctx) used by UI and template rendering (Email/Webhook/etc.). It also exposes helpers for working with nested dot paths and for compacting large step-context blobs.

Schema (top-level keys produced by build_context()):

• meta: {“schema”: int}

• workflow: {“id”: str, “name”: str, “instance_id”: Any, “instance_state”: str}

• device: {

  “common_name”: str, “serial_number”: str, “device_id”: Any, “domain”: Any, “device_type”: Any, “created_at”: Any,

  }

• request: {

  “protocol”: str | None, “operation”: str | None, “enrollment_request_id”: str | None, “csr_pem”: str | None, …CSR-derived fields from _parse_csr_info…

  }

• steps: dict[str, Any] safe keys usable with dot lookup (e.g. “step_2”)

• vars: dict[str, Any] merged global variables bucket ($vars)

Notes:

• Use {{ ctx.steps.step_1 }} in templates (recommended).

Attributes

CTX_SCHEMA_VERSION
VARS_MAX_BYTES
STEP_CTX_MAX_BYTES
STEP_TEXT_EXCERPT

Functions

get_in(root, path)	Return value at dot path from a nested dict or raise KeyError.
set_in(root, path, value, *[, forbid_overwrite])	Set value at dot path. If forbid_overwrite=True, raise on value change collisions.
build_context(instance)	Compose the template context ctx for a workflow instance.
compact_context_blob(blob)	Compact a step-context blob to fit STEP_CTX_MAX_BYTES.

Module Contents

workflows.services.context.CTX_SCHEMA_VERSION: int = 1

workflows.services.context.VARS_MAX_BYTES: int

workflows.services.context.STEP_CTX_MAX_BYTES: int

workflows.services.context.STEP_TEXT_EXCERPT: int

workflows.services.context.get_in(root, path)

Return value at dot path from a nested dict or raise KeyError.

Parameters:

• root (dict[str, Any]) – Root dictionary to traverse.

• path (str) – Dot-separated path (e.g. "a.b.c").

Returns:

The value found at the given path.

Return type:

Any

Raises:

• KeyError – If the full path does not exist in the nested dictionaries.

• ValueError – If the path is empty or contains illegal segments.

workflows.services.context.set_in(root, path, value, *, forbid_overwrite=True)

Set value at dot path. If forbid_overwrite=True, raise on value change collisions.

Parameters:

• root (dict[str, Any])

• path (str)

• value (Any)

• forbid_overwrite (bool)

Return type:

None

workflows.services.context.build_context(instance)

Compose the template context ctx for a workflow instance.

Parameters:

instance (workflows.models.WorkflowInstance) – Workflow instance for which to build the context.

Returns:

A plain dict suitable for Django templates.

Return type:

dict[str, Any]

workflows.services.context.compact_context_blob(blob)

Compact a step-context blob to fit STEP_CTX_MAX_BYTES.

The compaction is lossy: long strings are truncated, large nested dicts are summarized. A "_meta" key indicates truncation and original size.

Parameters:

blob (dict[str, Any]) – Original step-context dictionary.

Returns:

Either the original blob, or a compacted summary that fits within STEP_CTX_MAX_BYTES.

Return type:

dict[str, Any]