Workflow API¶
general_manager.workflow.event_registry.WorkflowEvent dataclass ¶
Canonical workflow trigger event routed by workflow event registries.
Attributes:
| Name | Type | Description |
|---|---|---|
event_id | str | Stable idempotency key for duplicate suppression. |
event_type | str | Canonical dotted type such as |
payload | WorkflowEventPayload | Event-specific payload stored in memory or serialized to the database-backed event record. |
event_name | str | None | Optional readable routing name such as |
source | str | None | Optional origin label for observability and auditing. |
occurred_at | datetime | None | Optional timestamp for when the domain event occurred. |
metadata | WorkflowEventPayload | Additional event metadata stored alongside the payload. |
The dataclass is frozen but does not validate field values at runtime. Empty strings and caller-supplied mutable mappings are accepted by the constructor. In-memory registries pass mappings through to handlers as-is; database registries snapshot them with dict(...) and rely on Django model fields for JSON/date serialization validation.
general_manager.workflow.event_registry.WorkflowEventHandler module-attribute ¶
WorkflowEventHandler = Callable[[WorkflowEvent], None]
general_manager.workflow.event_registry.EventValidator module-attribute ¶
EventValidator = Callable[[WorkflowEvent], None]
general_manager.workflow.event_registry.EventPredicate module-attribute ¶
EventPredicate = Callable[[WorkflowEvent], bool]
general_manager.workflow.event_registry.DeadLetterHandler module-attribute ¶
DeadLetterHandler = Callable[
[WorkflowEvent, Exception], None
]
general_manager.workflow.event_registry.RetryPredicate module-attribute ¶
RetryPredicate = Callable[[Exception], bool]
general_manager.workflow.events.manager_created_event ¶
manager_created_event(
*,
manager,
values,
identification=None,
event_name="manager_created",
event_id=None,
source=None,
occurred_at=None,
metadata=None
)
Build a workflow event for a manager create operation.
event_id defaults to a UUID4 string and occurred_at defaults to the current UTC time. The payload always contains manager and a shallow copy of values; identification is included only when provided. Metadata is shallow-copied into the resulting WorkflowEvent.
general_manager.workflow.events.manager_updated_event ¶
manager_updated_event(
*,
manager,
changes,
old_values=None,
identification=None,
event_name="manager_updated",
event_id=None,
source=None,
occurred_at=None,
metadata=None
)
Build a workflow event for a manager update operation.
The payload contains manager and changes, where each changed field maps to {"old": old_values.get(field), "new": value}. Missing old values are represented as None. identification is included only when provided. event_id, occurred_at, and metadata follow manager_created_event.
general_manager.workflow.events.manager_deleted_event ¶
manager_deleted_event(
*,
manager,
identification=None,
event_name="manager_deleted",
event_id=None,
source=None,
occurred_at=None,
metadata=None
)
Build a workflow event for a manager delete operation.
The payload always contains manager and includes identification only when provided. event_id, occurred_at, and metadata follow manager_created_event.
The manager event helpers are public constructors for common CRUD workflow events. They return WorkflowEvent instances with canonical event types: general_manager.manager.created, general_manager.manager.updated, and general_manager.manager.deleted. If omitted, event_id is generated as a UUID4 string and occurred_at is the current UTC time.
Payload contracts are stable:
manager_created_event(...)stores{"manager": name, "values": {...}}and includesidentificationonly when provided.manager_updated_event(...)stores{"manager": name, "changes": {...}}, where each changed field maps to{"old": old_values.get(field), "new": value}. Missing old values are represented asNone.manager_deleted_event(...)stores{"manager": name}and includesidentificationonly when provided.
The helpers shallow-copy top-level values, changes, old_values, identification, and metadata mappings. Nested objects are not deep-copied or JSON-normalized; database registries rely on Django field serialization.
general_manager.workflow.event_registry.EventRegistry ¶
Bases: Protocol
Registry that validates, deduplicates, and routes workflow events.
register ¶
register(
event,
*,
handler,
validator=None,
when=None,
retries=0,
retry_on=None,
dead_letter_handler=None
)
Register a handler for an event type or event name.
event values containing a dot route by WorkflowEvent.event_type; values without a dot route by WorkflowEvent.event_name. validator runs before when; validator errors and final handler failures are sent to the registration's dead-letter handler when configured. retries counts additional attempts after the first and is clamped to zero when negative.
The registry trusts the annotated input types. It does not validate non-callable handlers, validators, predicates, retry predicates, or dead-letter handlers at registration time; invalid callables fail when the route is evaluated. Non-string event keys or non-integer retry values may raise normal Python errors during registration.
Route keys are matched by exact string equality; they are not stripped, normalized, wildcarded, or prefix-matched. Empty strings are valid keys and match events whose event_name is None or "". A readable event_name containing a dot is treated as a type route key, so prefer dot-free event names.
Identical registrations are ignored rather than appended. Registration identity is derived from the event key, handler, validator, predicate, retry count, retry predicate, and dead-letter handler. Different routing options produce separate registrations and may invoke the same handler more than once. Callable identity uses each callable's __module__ + "." + __qualname__ when available and repr(...) otherwise; retry count identity uses the clamped retry count.
Matching handlers run in registration order within each route bucket. Type-route registrations run before name-route registrations when an event matches both. A registration becomes applicable only after its validator succeeds and its when predicate is absent or returns True; when=False registrations are skipped and do not affect publish() return values.
retry_on=None retries every handler exception until the attempt budget is exhausted. Validator failures are not retried. when predicate exceptions and dead-letter handler exceptions are not wrapped by the in-memory registry; database outbox processing records them as outbox failures. If a retry_on predicate raises, that exception propagates the same way and is not converted into a retry decision.
publish ¶
publish(event)
Publish an event and return whether at least one handler completed.
Implementations may suppress duplicate event.event_id values and may persist work before handlers run. A duplicate publish returns False. If at least one applicable handler completes and another fails, synchronous registries return True.
general_manager.workflow.event_registry.InMemoryEventRegistry ¶
Bases: _RoutingMixin
Thread-safe in-memory event registry suitable for local development.
The registry deduplicates published events by event_id using a bounded process-local cache. It routes matching handlers synchronously and isolates handler failures so one failing route does not stop later routes.
max_seen_event_ids is clamped to at least 1; when the cache is full, the oldest id is evicted. The id is marked seen before routing, so a handler failure does not make the event id publishable again. The dataclass is frozen, but payload and metadata may still reference mutable mappings.
Registration-level dead-letter handlers take precedence over the registry-level dead_letter_handler; the registry-level handler is used only when a failed registration does not define one.
The internal lock protects registration mutation, handler snapshot creation, and seen-id cache updates. Handlers run outside the lock. Concurrent publish() calls use the handler snapshot available when routing starts; concurrent register() calls may affect later publishes but not a snapshot already being routed.
general_manager.workflow.event_registry.DatabaseEventRegistry ¶
Bases: _RoutingMixin
DB-backed event registry for production event durability.
publish() persists a WorkflowEventRecord and WorkflowOutbox row. In async mode it schedules outbox processing after commit and returns False because handlers have not run yet; otherwise it processes the outbox row synchronously. Duplicate event_id values are suppressed by the event record uniqueness constraint; duplicate publish() calls return False without creating or routing a new outbox row. Database registries do not have a registry-level dead-letter handler; use registration-level handlers for custom dead-letter handling. The class accepts no constructor options.
Payload and metadata values are handed to Django's JSONField; they must be serializable by the configured Django/database JSON handling. occurred_at is handed to Django's DateTimeField without conversion in this layer, so timezone behavior follows the project's Django settings and database backend.
publish_sync ¶
publish_sync(event)
Persist event if possible and route matching handlers inline.
A duplicate event_id is not inserted again, but inline routing still runs against the provided event object. Non-duplicate persistence errors propagate and prevent routing. Successful persistence commits before inline routing begins. Returns True when at least one handler completes.
process_outbox_entry ¶
process_outbox_entry(outbox_id, *, claim_token=None)
Route one outbox row and finalize its delivery state.
Returns False for missing or already processed rows, stale or missing ownership claims, duplicate in-progress delivery attempts, handler failures, and finalize failures. Rows with only filtered-out handlers are marked processed and return False. Successful handler completion marks the row processed and returns True.
Handler failure increments the outbox attempt count. While attempts are below WORKFLOW_MAX_RETRIES, the row remains failed and becomes due again at now + WORKFLOW_RETRY_BACKOFF_SECONDS * attempts. When attempts reach WORKFLOW_MAX_RETRIES and dead letters are enabled, the row moves to dead_letter. Handler-level register(..., retries=...) controls delivery attempts inside one outbox processing call; outbox attempts control later processing calls. If at least one applicable handler succeeds and another handler fails during the same processing call, the failed handler still receives dead-letter handling but the outbox row is marked processed because the event had a successful route.
Dead letters are controlled by WORKFLOW_DEAD_LETTER_ENABLED, which defaults to True and is interpreted with bool(...). When disabled, rows that reach WORKFLOW_MAX_RETRIES remain failed and are scheduled again with backoff instead of moving to dead_letter. WORKFLOW_MAX_RETRIES is parsed with int(...), clamped to at least 0, defaults to 3 on missing or invalid values, and counts total failed outbox processing calls before dead-letter transition. WORKFLOW_RETRY_BACKOFF_SECONDS is parsed with int(...), clamped to at least 1, and defaults to 5 on missing or invalid values.
claim_outbox_batch ¶
claim_outbox_batch(*, batch_size=None)
Claim available outbox rows for async processing.
The method claims pending/failed rows whose available_at is due and stale claimed rows whose claim TTL has expired, using row locks and one fresh claim token for the returned batch. batch_size=None uses the configured outbox batch size. Returns (outbox_id, claim_token) pairs.
outbox_snapshot ¶
outbox_snapshot()
Return (pending_count, oldest_pending_age_seconds) for operations UI.
The age is 0.0 when no pending rows exist or when clock skew would otherwise produce a negative age.
general_manager.workflow.event_registry.InvalidWorkflowEventRegistryOptionsError ¶
Bases: TypeError
Raised when WORKFLOW_EVENT_REGISTRY mapping options are not a mapping.
general_manager.workflow.event_registry.InvalidWorkflowEventRegistryError ¶
Bases: TypeError
Raised when a workflow event registry setting cannot resolve to a registry.
general_manager.workflow.event_registry.configure_event_registry ¶
configure_event_registry(registry)
Set the process-local active workflow event registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry | EventRegistry | Registry instance used by | required |
general_manager.workflow.event_registry.configure_event_registry_from_settings ¶
configure_event_registry_from_settings(django_settings)
Configure the workflow event registry from Django settings.
GENERAL_MANAGER["WORKFLOW_EVENT_REGISTRY"] takes precedence over a top-level WORKFLOW_EVENT_REGISTRY setting, including explicit None to use the WORKFLOW_MODE default registry. Values may be:
Noneor missing to use the mode default registry.- An
EventRegistryinstance. - A dotted import path to an
EventRegistryinstance, class, or factory. - A zero-argument callable returning an
EventRegistry. - A mapping with
{"class": <path-or-callable>, "options": {...}}; options are passed as keyword arguments when constructing/calling the reference. Other mapping keys are ignored and options are not merged with any other settings.
WORKFLOW_MODE values are normalized by stripping whitespace and lowercasing. "production" selects DatabaseEventRegistry; "local" and any unrecognized value select InMemoryEventRegistry.
Import, factory, and constructor exceptions propagate.
Dotted import strings are resolved before classification. Imported classes are instantiated, imported registry instances are reused, and imported callables that are not already EventRegistry instances are called as factories.
Raises:
| Type | Description |
|---|---|
TypeError | If mapping |
general_manager.workflow.event_registry.get_event_registry ¶
get_event_registry()
Return the currently configured process-local workflow event registry.
Before explicit configuration, this is the import-time InMemoryEventRegistry instance. Call configure_event_registry_from_settings() to replace it with the registry selected by Django settings and WORKFLOW_MODE.
general_manager.workflow.event_registry.publish_sync ¶
publish_sync(event)
Publish an event synchronously against the configured registry.
If the active registry exposes publish_sync, that method is used so database-backed registries can persist the event and route handlers inline even when async delivery is enabled. Otherwise this falls back to registry.publish(event).
Returns:
| Name | Type | Description |
|---|---|---|
bool | bool |
|
general_manager.workflow.config.workflow_mode ¶
workflow_mode(django_settings=settings)
Return the configured workflow mode.
Nested GENERAL_MANAGER["WORKFLOW_MODE"] takes precedence over the top-level setting. Values are coerced with str(...).strip().lower(). Only "local" and "production" are accepted; all other values fall back to "local".
general_manager.workflow.config.workflow_async_enabled ¶
workflow_async_enabled(django_settings=settings)
Return whether workflow handlers should be dispatched asynchronously.
Explicit non-None nested or top-level WORKFLOW_ASYNC values are coerced with bool(...); omitted or None values fall back to production mode.
general_manager.workflow.config.workflow_beat_enabled ¶
workflow_beat_enabled(django_settings=settings)
Return whether workflow Celery Beat scheduling is enabled.
Explicit non-None nested or top-level WORKFLOW_BEAT_ENABLED values are coerced with bool(...); omitted or None values fall back to production mode.
general_manager.workflow.config.workflow_beat_outbox_interval_seconds ¶
workflow_beat_outbox_interval_seconds(
django_settings=settings,
)
Return the workflow outbox Beat interval, clamped to at least 1 second.
general_manager.workflow.config.workflow_beat_max_jitter_seconds ¶
workflow_beat_max_jitter_seconds(django_settings=settings)
Return the workflow Beat jitter window, clamped to at least 0 seconds.
general_manager.workflow.config.workflow_outbox_batch_size ¶
workflow_outbox_batch_size(django_settings=settings)
Return async outbox publish batch size, clamped to at least 1.
general_manager.workflow.config.workflow_outbox_process_chunk_size ¶
workflow_outbox_process_chunk_size(
django_settings=settings,
)
Return outbox processing chunk size, clamped to at least 1.
general_manager.workflow.config.workflow_outbox_claim_ttl_seconds ¶
workflow_outbox_claim_ttl_seconds(django_settings=settings)
Return outbox claim lease TTL, clamped to at least 1 second.
general_manager.workflow.config.workflow_max_retries ¶
workflow_max_retries(django_settings=settings)
Return max failed outbox processing calls before dead-lettering.
Values are parsed with int(...), clamped to at least 0, and default to 3 when missing or invalid.
general_manager.workflow.config.workflow_retry_backoff_seconds ¶
workflow_retry_backoff_seconds(django_settings=settings)
Return retry backoff multiplier in seconds, clamped to at least 1.
general_manager.workflow.config.workflow_dead_letter_enabled ¶
workflow_dead_letter_enabled(django_settings=settings)
Return whether max-retry outbox failures move to dead-letter status.
Nested or top-level WORKFLOW_DEAD_LETTER_ENABLED values are always coerced with bool(...); an explicit None therefore disables dead letters.
general_manager.workflow.config.workflow_delivery_running_timeout_seconds ¶
workflow_delivery_running_timeout_seconds(
django_settings=settings,
)
Return stale running delivery-attempt timeout, clamped to at least 1.
Workflow config helpers read nested GENERAL_MANAGER values before top-level Django settings. Non-mapping GENERAL_MANAGER values are ignored. Boolean helpers use Python bool(...) coercion for explicit non-None values; WORKFLOW_ASYNC and WORKFLOW_BEAT_ENABLED treat None as omitted and fall back to mode defaults. WORKFLOW_DEAD_LETTER_ENABLED always uses bool(...), so an explicit None disables dead letters. Integer helpers parse with int(...), clamp to their documented minimums, and return their defaults when parsing fails. workflow_mode() normalizes with str(...).strip().lower() and accepts only local or production. The helpers do not wrap unexpected errors raised while reading settings attributes.
general_manager.workflow.models.WorkflowEventRecord ¶
Bases: Model
Durable workflow event payload routed through the event registry.
general_manager.workflow.models.WorkflowOutbox ¶
Bases: Model
Outbox entry used to claim, retry, and route workflow events.
general_manager.workflow.models.WorkflowExecutionRecord ¶
Bases: Model
Durable workflow execution state used by production workflow engines.
The state column is stored as a raw database string. Workflow engines narrow it to the public WorkflowState vocabulary when returning WorkflowExecution DTOs or raising state-transition errors.
general_manager.workflow.models.WorkflowDeliveryAttempt ¶
Bases: Model
Per-handler delivery audit row keyed for idempotent event handling.
The workflow persistence models are durable implementation records used by the database event registry, outbox workers, and production workflow engines. WorkflowEventRecord stores the event payload and metadata, WorkflowOutbox tracks claim/retry/dead-letter routing state, WorkflowDeliveryAttempt records per-handler idempotent delivery status, and WorkflowExecutionRecord stores workflow execution state, input/output payloads, correlation ids, timestamps, errors, and metadata. Application code should normally use the registry and engine APIs instead of mutating these rows directly. Execution rows store state as raw database strings; engines narrow those values to the public WorkflowState vocabulary when returning WorkflowExecution snapshots or raising state-transition errors. Workflow model Meta.indexes use explicit names that match the checked-in migrations, including the initial workflow schema migration and later scaling-index migration. Treat those names as migration-owned metadata; renaming them creates an explicit index-rename migration.
general_manager.workflow.tasks.configure_workflow_beat_schedule_from_settings ¶
configure_workflow_beat_schedule_from_settings(
django_settings=settings,
)
Register the workflow outbox periodic drain schedule in Celery Beat.
WORKFLOW_BEAT_ENABLED controls whether scheduling is attempted. WORKFLOW_BEAT_OUTBOX_INTERVAL_SECONDS sets the base interval in seconds and defaults to 5; WORKFLOW_BEAT_MAX_JITTER_SECONDS defaults to 2 and adds a random fractional-second delay up to that value. The schedule entry is written to Celery's process-local current_app.conf.beat_schedule under WORKFLOW_BEAT_SCHEDULE_KEY.
Returns False when workflow beat is disabled, Celery is not installed, or Celery imported but current_app is None. When enabled, the function replaces/updates the process-local Celery beat_schedule entry for publish_outbox_batch and returns True. A False return leaves any existing schedule entry untouched.
The generated Beat entry has task name "general_manager.workflow.tasks.publish_outbox_batch", numeric schedule seconds, no args/kwargs, and options {"queue": "workflow.events"}. Jitter is sampled once when this function configures the entry, not on each Beat tick.
Celery task-layer retry behavior is not configured here; exceptions from Celery app access or schedule assignment propagate.
general_manager.workflow.tasks.publish_outbox_batch ¶
publish_outbox_batch()
Claim available workflow outbox entries and dispatch one batch route task.
Returns:
| Name | Type | Description |
|---|---|---|
int | int | Number of outbox rows claimed for routing. Returns |
int | active event registry is not a |
Side effects
Updates workflow outbox telemetry after claiming. When Celery is available, claimed rows are dispatched to route_outbox_claims_batch via .delay(...); otherwise they are routed inline.
WORKFLOW_OUTBOX_PROCESS_CHUNK_SIZE controls the claim batch size. DatabaseEventRegistry.claim_outbox_batch() defines claim ordering and claim TTL behavior. Empty claim batches still refresh outbox snapshot telemetry and do not enqueue a route task. Exceptions from claiming, telemetry, inline routing, or Celery .delay(...) propagate to the caller/Celery worker.
general_manager.workflow.tasks.route_outbox_event ¶
route_outbox_event(outbox_id, claim_token=None)
Route one workflow outbox row through the active database event registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
outbox_id | int | Primary key of the outbox row to process. | required |
claim_token | str | None | Optional ownership token from | None |
Returns:
| Name | Type | Description |
|---|---|---|
bool | bool |
|
bool | row processed. Returns | |
bool |
|
Missing rows, already processed rows, claim-token mismatches, claimed rows without a matching token, duplicate in-progress delivery attempts, rows with no applicable handlers, and route failures return False according to DatabaseEventRegistry.process_outbox_entry(). This single-row task does not catch exceptions raised by the registry.
general_manager.workflow.tasks.route_outbox_claims_batch ¶
route_outbox_claims_batch(claims)
Route a claimed outbox batch with per-entry failure isolation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
claims | list[tuple[int, str]] |
| required |
Returns:
| Name | Type | Description |
|---|---|---|
int | int | Count of rows whose route processing returned |
Side effects
Exceptions from individual rows are logged and do not stop later rows in the same batch.
The public contract expects a concrete list of (int, str) pairs. Malformed claim entries raise normal Python unpacking/type errors inside that entry, are logged, and are then skipped by the batch loop. Duplicate claim entries are processed in list order and rely on per-row registry ownership checks.
general_manager.workflow.tasks.execute_workflow_handler ¶
execute_workflow_handler(
execution_id, handler_path, input_data=None
)
Run a persisted workflow execution handler and store its terminal state.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
execution_id | str | Durable workflow execution id. | required |
handler_path | str | Dotted import path to the callable handler. | required |
input_data | WorkflowTaskPayload | None | Optional handler payload. It is copied into a mutable dict before the handler is called. | None |
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
Side effects
Pending executions move to running, then completed with handler output. Executions no longer in pending are left unchanged. Handler import errors and handler exceptions are captured on the execution record as a failed state.
The handler is a synchronous callable accepting one dict[str, object] payload and returning a mapping, None, or another falsey value. None input becomes {}. Successful output is stored with dict(result or {}), so non-mapping truthy outputs fail the execution. Missing execution ids raise WorkflowExecutionNotFoundError. The task returns None for skipped, completed, and failed execution paths; callers inspect the execution record to distinguish those outcomes.
Row locking and conditional updates prevent completed/failed/cancelled state from being overwritten by concurrent workers. This task does not configure Celery retries; WorkflowExecutionNotFoundError propagates, while handler import/call/output conversion failures are captured as failed executions.
general_manager.workflow.tasks.resume_execution_task ¶
resume_execution_task(execution_id, signal=None)
Mark a waiting persisted execution as completed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
execution_id | str | Durable workflow execution id. | required |
signal | WorkflowTaskPayload | None | Optional resume signal stored in execution metadata. | None |
Returns:
| Name | Type | Description |
|---|---|---|
bool | bool |
|
bool | execution exists but is not in |
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
The resume signal is stored only in execution metadata under resume_signal; output data is not changed. This task does not configure Celery retries, so unexpected persistence errors propagate to the caller/worker.
general_manager.workflow.tasks.cancel_execution_task ¶
cancel_execution_task(execution_id, reason=None)
Cancel an active persisted workflow execution.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
execution_id | str | Durable workflow execution id. | required |
reason | str | None | Optional cancellation reason stored as the execution error. | None |
Returns:
| Name | Type | Description |
|---|---|---|
bool | bool |
|
bool | execution exists but is already terminal. |
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
Active states are pending, running, and waiting. The optional reason is stored in the execution error field. This task does not configure Celery retries, so unexpected persistence errors propagate to the caller/worker.
general_manager.workflow.telemetry.set_outbox_snapshot ¶
set_outbox_snapshot(
*, pending_count, oldest_pending_age_seconds
)
Set the point-in-time workflow outbox backlog gauges.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pending_count | int | Current number of pending outbox rows. | required |
oldest_pending_age_seconds | float | Age in seconds of the oldest pending row. | required |
Negative values are clamped to 0/0.0 before recording. The helper is a no-op when prometheus_client is unavailable. It does not catch exceptions raised by the installed metrics backend.
general_manager.workflow.telemetry.observe_outbox_claim_batch ¶
observe_outbox_claim_batch(size)
Observe the number of workflow outbox rows claimed in one batch.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
size | int | Claimed row count. Negative values are recorded as | required |
The helper is a no-op when prometheus_client is unavailable and does not catch exceptions raised by the installed metrics backend.
Raises:
| Type | Description |
|---|---|
TypeError | If |
general_manager.workflow.telemetry.observe_outbox_process_duration ¶
observe_outbox_process_duration(
*, status, duration_seconds
)
Observe workflow outbox row processing latency for a terminal status.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
status | str | Outbox status label. Spaces and colons are replaced with | required |
duration_seconds | float | Processing duration in seconds. Negative values are recorded as | required |
The helper is a no-op when prometheus_client is unavailable and does not catch exceptions raised by the installed metrics backend.
Raises:
| Type | Description |
|---|---|
TypeError | If |
general_manager.workflow.telemetry.increment_outbox_status ¶
increment_outbox_status(status)
Increment the workflow outbox status-transition counter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
status | str | Outbox status label. Spaces and colons are replaced with | required |
The helper is a no-op when prometheus_client is unavailable and does not catch exceptions raised by the installed metrics backend.
Raises:
| Type | Description |
|---|---|
TypeError | If |
general_manager.workflow.telemetry.increment_delivery_attempt ¶
increment_delivery_attempt(*, status)
Increment the workflow delivery-attempt status counter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
status | str | Delivery-attempt status label. Spaces and colons are replaced with | required |
The helper is a no-op when prometheus_client is unavailable and does not catch exceptions raised by the installed metrics backend.
Raises:
| Type | Description |
|---|---|
TypeError | If |
general_manager.workflow.telemetry.increment_execution_state ¶
increment_execution_state(state)
Increment the workflow execution state counter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state | str | Execution state label. Spaces and colons are replaced with | required |
The helper is a no-op when prometheus_client is unavailable and does not catch exceptions raised by the installed metrics backend.
general_manager.workflow.telemetry.increment_duplicate_suppression ¶
increment_duplicate_suppression()
Increment the suppressed duplicate workflow delivery-attempt counter.
The helper is a no-op when prometheus_client is unavailable and does not catch exceptions raised by the installed metrics backend.
general_manager.workflow.telemetry.extract_outbox_snapshot_payload ¶
extract_outbox_snapshot_payload(snapshot)
Extract typed outbox snapshot values from a serialized mapping.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
snapshot | Mapping[str, object] | Mapping that may contain | required |
Returns:
| Type | Description |
|---|---|
int | tuple[int, float]: Pending count and oldest pending age. Missing or |
float | falsey values are returned as |
tuple[int, float] | Python truth-value testing, so |
tuple[int, float] | strings/bytes, and empty containers are treated as absent; exceptions |
tuple[int, float] | from custom truth-value methods propagate. Parsed negative values are |
tuple[int, float] | returned unchanged; recording helpers clamp when they emit gauges or |
tuple[int, float] | observations. |
Coercion
pending_count is parsed with int(...), so floats are truncated, True becomes 1, "1.2" raises ValueError, and non-finite floats raise the exception that int(...) raises. oldest_pending_age_seconds is parsed with float(...), so "nan"/"inf" and non-finite floats are accepted by Python's normal float conversion rules.
Raises:
| Type | Description |
|---|---|
TypeError | If a present truthy value is not coercible to the expected numeric type, including errors raised by the underlying conversion. |
ValueError | Propagated from |
OverflowError | Propagated from |
general_manager.workflow.engine.WorkflowDefinition dataclass ¶
Workflow declaration metadata.
workflow_id is the durable backend identity. handler receives a workflow input mapping and may return an output mapping or None. Metadata is stored as provided by callers; the dataclass is frozen but does not deep-freeze mapping contents. The dataclass performs no custom runtime validation.
general_manager.workflow.engine.WorkflowExecution dataclass ¶
Execution snapshot returned by workflow engines.
state is typed as one of the WorkflowState literals for static checking. The dataclass performs no custom runtime validation. input_data, output_data, and metadata are stored exactly as provided by the backend; the dataclass is frozen but nested mapping contents are not copied or deep-frozen.
general_manager.workflow.engine.WorkflowEngine ¶
Bases: Protocol
Protocol for workflow orchestration backends.
Implementations define their own persistence, copying, handler execution, retry, and error-capture behavior, but they return WorkflowExecution snapshots and use the shared workflow state vocabulary.
general_manager.workflow.engine.WorkflowState module-attribute ¶
WorkflowState = Literal[
"pending",
"running",
"waiting",
"failed",
"cancelled",
"completed",
]
general_manager.workflow.engine.WorkflowEngineError ¶
Bases: RuntimeError
Base class for workflow engine failures.
general_manager.workflow.engine.WorkflowExecutionNotFoundError ¶
Bases: WorkflowEngineError
Raised when an execution id cannot be resolved.
The error message is Workflow execution '<id>' was not found..
general_manager.workflow.engine.WorkflowCancelledError ¶
Bases: WorkflowEngineError
Raised when an operation targets a cancelled workflow.
The error message is Workflow execution '<id>' is cancelled..
general_manager.workflow.engine.WorkflowInvalidStateError ¶
Bases: WorkflowEngineError
Raised when an operation is not valid for the current workflow state.
The error message includes the execution id, attempted operation, current state, and a comma-joined expected state list in the format Workflow execution '<id>' cannot <operation> from state '<state>'. Expected one of: <states>..
WorkflowDefinition and WorkflowExecution are frozen dataclasses shared by all workflow backends. Payload fields use Mapping[str, object]: the dataclass itself is immutable, but mappings are stored exactly as provided and nested contents are not copied, deep-frozen, or JSON-normalized. The dataclasses perform no custom runtime validation beyond normal Python construction; WorkflowState is the static public state vocabulary. WorkflowDefinition.workflow_id is the backend workflow identity, and an optional handler receives input data and returns a mapping or None. The shared state tuples group active, active-or-completed, and terminal states for backend validation. Workflow engine errors include the execution id. Not-found and cancelled errors use Workflow execution '<id>' was not found. and Workflow execution '<id>' is cancelled.; invalid-state errors include the attempted operation, current state, and comma-joined expected state list.
general_manager.workflow.actions.Action ¶
Bases: Protocol
Executable workflow action used by ActionRegistry.
Actions receive read-only context and parameter mappings and either return a mapping result or None. Exceptions raised by the implementation are wrapped by ActionRegistry.execute().
general_manager.workflow.actions.ActionAlreadyRegisteredError ¶
Bases: ActionError
Raised when an action is registered more than once.
The error message includes the duplicate action name.
general_manager.workflow.actions.ActionExecutionError ¶
Bases: ActionError
Raised when action execution fails.
The error message includes the action name.
general_manager.workflow.actions.ActionNotFoundError ¶
Bases: ActionError
Raised when an action name is not registered.
The error message includes the missing action name.
general_manager.workflow.actions.ActionRegistry ¶
Process-local in-memory registry for exact action name strings.
The registry stores actions without runtime protocol validation. Invalid action objects fail later when execute() calls their execute attribute.
register ¶
register(name, action, *, replace=False)
Register action under the exact string name.
Raises:
| Type | Description |
|---|---|
ActionAlreadyRegisteredError | If |
get ¶
get(name)
Return the action registered as the exact string name.
Raises:
| Type | Description |
|---|---|
ActionNotFoundError | If no action has been registered for |
execute ¶
execute(name, *, context=None, params=None)
Execute the named action with optional context and params.
Missing context or params are passed as fresh empty dictionaries. Supplied mappings are passed through unchanged, including falsey mappings. Lookup failures raise ActionNotFoundError; exceptions from the action itself are wrapped in ActionExecutionError with the original exception as the cause.
Workflow actions are named side-effect adapters invoked from workflow handlers. An Action implements execute(context, params) and returns a mapping result or None. ActionRegistry is process-local and matches exact action name strings; names are not normalized. register(name, action) stores the action without runtime protocol validation and rejects duplicate names with ActionAlreadyRegisteredError unless replace=True is passed. get(name) raises ActionNotFoundError for missing names. execute(name, ...) applies fresh empty dictionaries only when context or params are omitted, preserves supplied mapping objects including falsey mappings, and wraps exceptions raised by the action in ActionExecutionError. Action error messages include the relevant action name. names() returns registered names sorted alphabetically.
general_manager.workflow.backend_registry.configure_workflow_engine ¶
configure_workflow_engine(engine)
Set the process-local active workflow engine.
Pass None to clear the configured engine so the next get_workflow_engine() call reads settings and may install the WORKFLOW_MODE default.
general_manager.workflow.backend_registry.configure_workflow_engine_from_settings ¶
configure_workflow_engine_from_settings(django_settings)
Configure the workflow engine from Django settings.
GENERAL_MANAGER["WORKFLOW_ENGINE"] takes precedence over a top-level WORKFLOW_ENGINE setting, including explicit None to clear the configured engine and allow get_workflow_engine() to use the WORKFLOW_MODE default. Values may be:
Noneor missing to clear the active engine.- A
WorkflowEngineinstance. - A dotted import path to a
WorkflowEngineinstance, class, or factory. - A zero-argument callable returning a
WorkflowEngine. - A mapping with
{"class": <path-or-callable>, "options": {...}}; options are passed as keyword arguments when constructing/calling the reference.
Import, factory, and constructor exceptions propagate.
Raises:
| Type | Description |
|---|---|
TypeError | If mapping |
general_manager.workflow.backend_registry.get_workflow_engine ¶
get_workflow_engine()
Return the configured workflow engine, installing settings/defaults first.
If no process-local engine is active, this function reads Django settings via configure_workflow_engine_from_settings(django.conf.settings). If settings leave the engine unset, production WORKFLOW_MODE installs one CeleryWorkflowEngine; all other modes install one LocalWorkflowEngine. The installed default is cached for later calls.
general_manager.workflow.signal_bridge.configure_workflow_signal_bridge_from_settings ¶
configure_workflow_signal_bridge_from_settings(
django_settings,
)
Connect or disconnect the bridge based on Django settings.
general_manager.workflow.signal_bridge.connect_workflow_signal_bridge ¶
connect_workflow_signal_bridge(*, registry=None)
Connect manager mutation signal bridging into workflow events.
If registry is provided, it becomes the active global registry before the receiver is connected. The receiver is connected with a stable dispatch uid and weak=False, so repeated calls replace the same receiver registration.
general_manager.workflow.signal_bridge.disconnect_workflow_signal_bridge ¶
disconnect_workflow_signal_bridge()
Disconnect the workflow signal bridge receiver by dispatch uid.
The signal bridge connects general_manager.cache.signals.post_data_change to workflow publishing. connect_workflow_signal_bridge(registry=...) optionally installs the provided event registry, then connects a receiver with a stable dispatch uid and weak=False; disconnect_workflow_signal_bridge() removes that receiver by dispatch uid. workflow_signal_bridge_enabled(settings) reads nested GENERAL_MANAGER["WORKFLOW_SIGNAL_BRIDGE"] before the top-level setting, ignores non-mapping GENERAL_MANAGER values, uses bool(...) coercion, and defaults to False. The receiver ignores the Django signal sender and uses instance when present, otherwise previous_instance from signal kwargs.
The bridge publishes create, update, and delete events only for GeneralManager instances; unknown actions are ignored. Create/update events require at least one non-reserved changed field. Event payloads use the manager class name, identification from the signal payload or current manager identification, source general_manager.cache.signals.post_data_change, and metadata {"action": action}. Event ids and timestamps use the manager event helper defaults. Update events start with old_relevant_values; when an old value is missing or None, the bridge uses a non-None value from the previous simple-history row when available. Exceptions from publish() propagate to the signal caller.
general_manager.workflow.backends.local.LocalWorkflowEngine ¶
Process-local workflow engine for development and tests.
Executions are stored in memory, so state is lost when the process exits. Input data, metadata, handler input, and resume signals are deep-copied at the engine boundary to isolate stored snapshots from caller mutation.
start ¶
start(
workflow,
input_data=None,
*,
correlation_id=None,
metadata=None
)
Start workflow and return an execution snapshot.
Workflows without handlers complete with an empty output mapping. Handler exceptions are captured as failed executions. A non-empty correlation_id reuses an existing active or completed local execution for the same workflow id. Concurrent starts for the same correlation key wait for the in-flight start to finish and then return that snapshot. Failed or cancelled snapshots do not block a fresh start.
resume ¶
resume(execution_id, signal=None)
Complete a waiting execution with an optional resume signal.
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
WorkflowCancelledError | If the execution is already cancelled. |
WorkflowInvalidStateError | If the execution is not waiting. |
cancel ¶
cancel(execution_id, *, reason=None)
Cancel an active local execution.
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
WorkflowCancelledError | If the execution is already cancelled. |
WorkflowInvalidStateError | If the execution is terminal. |
status ¶
status(execution_id)
Return the stored execution snapshot.
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
LocalWorkflowEngine is a process-local backend for development and tests. start(...) deep-copies input data and metadata into the returned WorkflowExecution, deep-copies handler input, completes immediately when no handler is configured, and records handler exceptions as failed executions. A non-empty correlation_id reuses the existing local execution for the same workflow id, including failed executions. Concurrent starts with the same local correlation key wait for the in-flight start and return its completed or failed snapshot. resume(...) only accepts waiting executions, stores a deep copy of the supplied signal under metadata["resume_signal"] when a signal is provided, and completes the execution. cancel(...) only accepts active executions and stores the optional reason as WorkflowExecution.error. status(...), resume(...), and cancel(...) raise WorkflowExecutionNotFoundError for unknown ids; invalid states raise the documented workflow state errors. Stored state is in memory only and is not shared across processes.
general_manager.workflow.backends.celery.CeleryWorkflowEngine ¶
Durable workflow engine backed by WorkflowExecutionRecord rows.
The engine stores every execution in the database. In sync mode it runs importable or inline handlers in the caller process. In async mode it creates a pending execution and schedules execute_workflow_handler with Celery after the surrounding transaction commits.
Returned WorkflowExecution objects are snapshots. In async mode, the Celery worker may update the execution immediately after start() returns; call status() when fresh state is required.
start ¶
start(
workflow,
input_data=None,
*,
correlation_id=None,
metadata=None
)
Persist and start one workflow execution.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
workflow | WorkflowDefinition | Workflow definition to execute. | required |
input_data | WorkflowPayload | None | Optional JSON-like payload copied with | None |
correlation_id | str | None | Optional durable dedupe key scoped to | None |
metadata | WorkflowPayload | None | Optional metadata copied with | None |
Returns:
| Name | Type | Description |
|---|---|---|
WorkflowExecution | WorkflowExecution | The new execution, or an existing active or |
WorkflowExecution | completed execution with the same |
Side effects
Creates a WorkflowExecutionRecord, records execution-state telemetry, and in async mode schedules the Celery handler task after transaction commit. In sync mode, handler output is stored inline.
In async mode, handlers must resolve to importable top-level callables. A workflow with neither workflow.handler nor metadata["handler_path"] is intentionally handlerless and completes immediately with {} output. A workflow with an inline/local handler that cannot be represented as a handler path is malformed for async mode and is recorded as failed. Non-callable imports, import failures, and missing Celery support also create failed execution records instead of raising. Runtime async handler import/call/output failures are captured later by execute_workflow_handler as failed execution records. In sync mode, handler exceptions and invalid truthy handler outputs are captured as failed executions.
workflow.workflow_id is the durable workflow identity. When an existing active or completed execution is reused for a correlation id, the new input_data and metadata are ignored. Failed executions are not reused, so the same correlation id may start another attempt. User metadata is copied before persistence; when a handler path is available, the engine writes/overwrites metadata["handler_path"] with the import path used for dispatch. Payload and metadata keys must be strings at the top level because the public type is Mapping[str, object]; nested values are accepted or rejected by the configured Django JSONField. Serialization failures propagate. The returned object is a snapshot of the row after creation and any inline sync handler execution, not a live object.
Raises:
| Type | Description |
|---|---|
IntegrityError | If the execution insert fails and no reusable correlation execution can be found. |
Exception | Propagates unexpected database, transaction, telemetry, or Celery task scheduling errors. |
resume ¶
resume(execution_id, signal=None)
Complete a waiting workflow execution with an optional resume signal.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
execution_id | str | Durable execution id. | required |
signal | WorkflowPayload | None | Optional JSON-like payload copied into execution metadata under | None |
Returns:
| Name | Type | Description |
|---|---|---|
WorkflowExecution | WorkflowExecution | The completed execution record. |
The update is applied inline and immediately changes the persisted state from waiting to completed; it does not enqueue a Celery task. A signal=None leaves metadata unchanged. Any explicit mapping, including an empty one, is copied into metadata["resume_signal"]. Output data is not changed.
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
WorkflowCancelledError | If the execution is already |
WorkflowInvalidStateError | If the execution is |
Exception | Propagates unexpected database or transaction errors. |
cancel ¶
cancel(execution_id, *, reason=None)
Cancel an active workflow execution.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
execution_id | str | Durable execution id. | required |
reason | str | None | Optional cancellation reason stored as the execution error. | None |
Returns:
| Name | Type | Description |
|---|---|---|
WorkflowExecution | WorkflowExecution | The cancelled execution record. |
The update is applied inline and immediately changes the persisted state to cancelled; it does not enqueue a Celery task. reason is stored in the execution error field and returned as WorkflowExecution.error.
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
WorkflowCancelledError | If the execution is already cancelled. |
WorkflowInvalidStateError | If the execution is |
Exception | Propagates unexpected database or transaction errors. |
status ¶
status(execution_id)
Return the current persisted execution state.
The database record stores state as a raw string; the returned WorkflowExecution snapshot exposes that value narrowed to the public WorkflowState vocabulary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
execution_id | str | Durable execution id. | required |
Returns:
| Name | Type | Description |
|---|---|---|
WorkflowExecution | WorkflowExecution | Snapshot of the stored execution record. |
Raises:
| Type | Description |
|---|---|
WorkflowExecutionNotFoundError | If |
Exception | Propagates unexpected database errors. |
general_manager.workflow.backends.n8n.N8nWorkflowEngine ¶
Configuration holder for the future n8n workflow adapter.
The constructor stores base_url and optional api_key, but no network operations are implemented yet. start(), resume(), cancel(), and status() all raise N8nOperationNotImplementedError.
__init__ ¶
__init__(*, base_url, api_key=None)
Store the n8n endpoint configuration for future adapter work.
start ¶
start(
workflow,
input_data=None,
*,
correlation_id=None,
metadata=None
)
Raise because starting n8n workflows is not implemented.
cancel ¶
cancel(execution_id, *, reason=None)
Raise because cancelling n8n workflows is not implemented.
N8nWorkflowEngine is a placeholder adapter for future remote orchestration. The constructor stores base_url and optional api_key, but start(...), resume(...), cancel(...), and status(...) are not implemented and raise N8nOperationNotImplementedError. The current adapter performs no network requests.