angee.workflows.models
Source models for workflow definitions.
The workflows addon owns graph definitions as data: a draft workflow lineage head carries editable steps, edges, and triggers, while publish() copies that draft into an immutable version. Step behavior remains in registry-selected StepImpl classes, so row data names keys and config, not Python callables. Future runtime subject/artifact references use Django contenttypes-backed object references; public ids stay at the transport boundary.
WorkflowStatus
class WorkflowStatus(models.TextChoices)Publication lifecycle for a workflow definition row.
JoinRule
class JoinRule(models.TextChoices)How a step with multiple incoming edges activates over upstream siblings.
TriggerKind
class TriggerKind(models.TextChoices)How a workflow lineage is started.
RunStatus
class RunStatus(models.TextChoices)Execution lifecycle for one pinned workflow run.
StepRunStatus
class StepRunStatus(models.TextChoices)Execution lifecycle for one step-run journal row.
Verdict
class Verdict(models.TextChoices)Resolution lifecycle for one awaited decision slot.
WorkflowQuerySet
class WorkflowQuerySet(AngeeQuerySet[Any])QuerySet owning subject declaration discovery and version currency.
current_published
def current_published() -> SelfReturn rows that are the current published version of their lineage.
The currency rule's single owner: a row survives when it is PUBLISHED and no _CURRENCY_STATUSES sibling in the same lineage is newer by (version, pk) — so a newer ARCHIVED row retires the lineage.
for_subject_declaration
def for_subject_declaration(subject_declaration: str) -> SelfReturn current published workflows accepting subject_declaration.
WorkflowManager
class WorkflowManager(AngeeManager.from_queryset(WorkflowQuerySet))Manager owning workflow lineage lookups.
current_published_for
def current_published_for(workflow: Any) -> Any | NoneReturn the latest published version for workflow's lineage.
Composes the same _CURRENCY_STATUSES rule current_published owns, scoped to one explicit lineage pool.
WorkflowRunQuerySet
class WorkflowRunQuerySet(AngeeQuerySet[Any])QuerySet owning workflow-run subject lookups.
for_subject
def for_subject(subject: Any) -> SelfReturn runs whose generic subject is subject.
WorkflowRunManager
class WorkflowRunManager(AngeeManager.from_queryset(WorkflowRunQuerySet))Manager owning workflow-run subject lookups.
Workflow
class Workflow(AuditMixin, AngeeDataModel)Editable workflow lineage head or immutable published workflow version.
A resource-assigned stable key identifies the lineage independently of its mutable display name and is shared by every published version.
Meta
class Meta()Django model options for workflow definitions.
__str__
def __str__() -> strReturn the workflow's display label.
after_resource_load
@classmethod
def after_resource_load(cls,
instances: Iterable[Any],
*,
tier: str,
source: str,
publish: bool = False) -> NoneReconcile stable keys and publish loaded drafts when requested.
mark_published
@transition(status,
source=WorkflowStatus.DRAFT,
target=WorkflowStatus.PUBLISHED,
on_success=_save_workflow_status)
def mark_published() -> NoneMark this copied version as published.
archive
@transition(
status,
source=WorkflowStatus.PUBLISHED,
target=WorkflowStatus.ARCHIVED,
on_success=_save_workflow_status,
)
def archive() -> NoneArchive a published workflow version.
clean
def clean() -> NoneValidate lineage-owned links and normalize stable and subject keys.
validate_subject_declaration
def validate_subject_declaration(subject: Any) -> NoneRaise when subject does not satisfy this workflow's subject declaration.
save
def save(*args: Any, **kwargs: Any) -> NonePersist the workflow after enforcing immutability and model validation.
delete
def delete(*args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]Delete only mutable workflow rows.
publish
def publish() -> SelfCopy this draft lineage head into an immutable published version.
publish_if_changed
def publish_if_changed() -> Self | NonePublish this draft only when no current version has the same definition.
is_immutable
@property
def is_immutable() -> boolReturn whether this workflow version rejects definition edits.
Step
class Step(ImplDefaultsMixin, AuditMixin, AngeeDataModel)One node in a workflow definition graph.
Meta
class Meta()Django model options for workflow steps.
__str__
def __str__() -> strReturn the step's display label.
clean
def clean() -> NoneValidate the step implementation key and config.
save
def save(*args: Any, **kwargs: Any) -> NonePersist the step after enforcing parent immutability and validation.
delete
def delete(*args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]Delete only steps belonging to mutable workflow rows.
Edge
class Edge(AuditMixin, AngeeDataModel)Directed edge between two workflow steps.
Meta
class Meta()Django model options for workflow edges.
__str__
def __str__() -> strReturn a compact edge label.
clean
def clean() -> NoneValidate that an edge is fully contained in one workflow.
save
def save(*args: Any, **kwargs: Any) -> NonePersist the edge after enforcing parent immutability and validation.
delete
def delete(*args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]Delete only edges belonging to mutable workflow rows.
TriggerManager
class TriggerManager(AngeeManager)Manager owning trigger row claims and due schedule priming.
claim_due_event
def claim_due_event(trigger_id: int, *, timestamp: datetime) -> Any | NoneLock and record one enabled event trigger fire if rate limits allow it.
claim_due_schedule
def claim_due_schedule(trigger_id: int, *,
timestamp: datetime) -> tuple[Any, datetime] | NoneLock and advance one due schedule trigger if rate limits allow it.
prime_due_schedules
def prime_due_schedules(*, timestamp: datetime) -> intPersist initial fire times for enabled schedules missing next_fire_at.
check_event_trigger_change_publishers
def check_event_trigger_change_publishers(
app_configs: list[object] | None = None,
**kwargs: object) -> list[checks.CheckMessage]Report persisted event triggers targeting models outside the change feed.
Trigger
class Trigger(AuditMixin, AngeeDataModel)Start rule attached to a workflow lineage head.
Event triggers consume the GraphQL change feed: their target model must declare changes() so publisher wiring and workflow delivery agree.
Meta
class Meta()Django model options for workflow triggers.
__str__
def __str__() -> strReturn the trigger's display label.
clean
def clean() -> NoneValidate lineage ownership and trigger declaration shape.
save
def save(*args: Any, **kwargs: Any) -> NonePersist the trigger after model validation.
enable
def enable() -> NoneEnable this trigger through the model owner.
disable
def disable() -> NoneDisable this trigger through the model owner.
rate_limit_allows
def rate_limit_allows(*, timestamp: datetime) -> boolReturn whether this trigger can fire at timestamp.
record_fire
def record_fire(
*, timestamp: datetime, extra_update_fields: Iterable[str] = ()) -> NoneRecord one trigger fire and persist rate-limit counters.
condition_matches
def condition_matches(sender: type[models.Model],
instance: models.Model) -> boolReturn whether this event trigger matches a saved model instance.
initial_fire_at
def initial_fire_at(*, now: datetime) -> datetime | NoneReturn the first persisted due timestamp for this schedule trigger.
compute_next_fire_at
def compute_next_fire_at(*, after: datetime, now: datetime) -> datetime | NoneReturn the next scheduled occurrence after after and later than now.
config_mapping
@property
def config_mapping() -> Mapping[str, Any]Return trigger config when it is a JSON object.
WorkflowRun
class WorkflowRun(AuditMixin, RecordRefMixin, AngeeDataModel)One execution of a pinned published workflow version.
Meta
class Meta()Django model options for workflow runs.
is_terminal
@property
def is_terminal() -> boolReturn whether this run has reached a terminal status.
awaiting_decision
def awaiting_decision() -> boolReturn whether this run has an unresolved workflow decision.
mark_running
@transition(status,
source=RunStatus.PENDING,
target=RunStatus.RUNNING,
on_success=save_state)
def mark_running() -> NoneMark a pending run as actively orchestrating.
resume
@transition(status,
source=RunStatus.WAITING,
target=RunStatus.RUNNING,
on_success=save_state)
def resume() -> NoneMark a waiting run as actively orchestrating again.
mark_waiting
@transition(
status,
source=RunStatus.RUNNING,
target=RunStatus.WAITING,
on_success=save_state,
)
def mark_waiting(*, wake_at: Any = None) -> NoneMark a run as waiting on durable external or timer state.
mark_succeeded
@transition(
status,
source=[RunStatus.RUNNING, RunStatus.WAITING],
target=RunStatus.SUCCEEDED,
on_success=save_state,
)
def mark_succeeded() -> NoneMark a run as successful.
mark_failed
@transition(
status,
source=[RunStatus.PENDING, RunStatus.RUNNING, RunStatus.WAITING],
target=RunStatus.FAILED,
on_success=save_state,
)
def mark_failed(error: str = "") -> NoneMark a run as failed with an optional durable error message.
mark_canceled
@transition(
status,
source=[RunStatus.PENDING, RunStatus.RUNNING, RunStatus.WAITING],
target=RunStatus.CANCELED,
on_success=save_state,
)
def mark_canceled() -> NoneMark a run as canceled.
save
def save(*args: Any, **kwargs: Any) -> NonePersist the run while keeping trigger dedup keys immutable.
from_db
@classmethod
def from_db(cls, db: str | None, field_names: list[str],
values: list[Any]) -> SelfCapture immutable loaded facts without a save-time SELECT.
debit_budget
def debit_budget(delta: Mapping[str, int]) -> NoneAtomically add usage deltas to this run's budget ledger.
StepRun
class StepRun(AuditMixin, AngeeDataModel)Journal row for one workflow step execution or system-injected event.
Meta
class Meta()Django model options for workflow step-run journal rows.
is_terminal
@property
def is_terminal() -> boolReturn whether this journal row has reached a terminal status.
mark_started
@transition(
status,
source=[StepRunStatus.SCHEDULED, StepRunStatus.WAITING],
target=StepRunStatus.STARTED,
on_success=save_state,
)
def mark_started(*,
heartbeat_at: Any = None,
claimed_deliveries: int = 0) -> NoneClaim this row for execution.
record_attempt
def record_attempt(*, heartbeat_at: Any = None) -> NoneRecord one implementation invocation for this started row.
mark_waiting
@transition(status,
source=StepRunStatus.STARTED,
target=StepRunStatus.WAITING,
on_success=save_state)
def mark_waiting(*,
until: Any = None,
resume_state: dict[str, Any] | None = None) -> NonePersist durable wait conditions for this row.
wake
def wake(*, at: datetime) -> NoneMake this waiting journal row due without changing its state.
Event delivery changes only the durable due time. The engine owns the later WAITING → STARTED claim and therefore remains the sole scheduler of implementation work.
mark_succeeded
@transition(
status,
source=[StepRunStatus.STARTED, StepRunStatus.WAITING],
target=StepRunStatus.SUCCEEDED,
on_success=save_state,
)
def mark_succeeded(*, output: Any = None, outcome: str = "") -> NonePersist a successful step result.
mark_failed
@transition(
status,
source=[StepRunStatus.STARTED, StepRunStatus.WAITING],
target=StepRunStatus.FAILED,
on_success=save_state,
)
def mark_failed(*,
error: str = "",
stacktrace: str = "",
outcome: str = "failed") -> NonePersist a failed step result.
mark_skipped
@transition(
status,
source=[StepRunStatus.SCHEDULED, StepRunStatus.WAITING],
target=StepRunStatus.SKIPPED,
on_success=save_state,
)
def mark_skipped() -> NoneMark this row as skipped by routing or join semantics.
mark_canceled
@transition(
status,
source=[
StepRunStatus.SCHEDULED, StepRunStatus.STARTED, StepRunStatus.WAITING
],
target=StepRunStatus.CANCELED,
on_success=save_state,
)
def mark_canceled() -> NoneMark this row as canceled.
reschedule_for_override
@transition(
status,
source=[
StepRunStatus.SUCCEEDED, StepRunStatus.FAILED, StepRunStatus.CANCELED,
StepRunStatus.SKIPPED
],
target=StepRunStatus.SCHEDULED,
on_success=save_state,
)
def reschedule_for_override(*, input: Any = None) -> NoneReset a terminal journal row so a manual override can run it again.
Decision
class Decision(AuditMixin, AngeeDataModel)One awaited resolution slot for a suspended step-run.
Meta
class Meta()Django model options for workflow decisions.
is_terminal
@property
def is_terminal() -> boolReturn whether this decision has a terminal verdict.
form_schema_annotation
@classmethod
def form_schema_annotation(cls) -> dict[str, Any]Return the narrow ORM projection consumed by :attr:form_schema.
form_schema
@property
def form_schema() -> dict[str, Any] | NoneReturn the enforced JSON-authored form schema, excluding Python model schemas.
mark_completed
@transition(verdict,
source=Verdict.PENDING,
target=Verdict.COMPLETED,
on_success=save_state)
def mark_completed(*, resolution: Any = None, resolved_by: str = "") -> NoneResolve this slot as completed.
mark_rejected
@transition(verdict,
source=Verdict.PENDING,
target=Verdict.REJECTED,
on_success=save_state)
def mark_rejected(*, resolution: Any = None, resolved_by: str = "") -> NoneResolve this slot as rejected.
mark_escalated
@transition(verdict,
source=Verdict.PENDING,
target=Verdict.ESCALATED,
on_success=save_state)
def mark_escalated(*, resolution: Any = None, resolved_by: str = "") -> NoneResolve this slot as escalated.
mark_expired
@transition(verdict,
source=Verdict.PENDING,
target=Verdict.EXPIRED,
on_success=save_state)
def mark_expired(*, resolution: Any = None, resolved_by: str = "") -> NoneResolve this slot as expired.
record_invalid_resolution
def record_invalid_resolution() -> NoneRecord one failed validation attempt while leaving the slot pending.
resolve
def resolve(verdict: Verdict,
*,
resolution: Any = None,
resolved_by: str = "") -> NoneResolve this slot through the transition matching verdict.