Skip to content

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

python
class WorkflowStatus(models.TextChoices)

Publication lifecycle for a workflow definition row.

JoinRule

python
class JoinRule(models.TextChoices)

How a step with multiple incoming edges activates over upstream siblings.

TriggerKind

python
class TriggerKind(models.TextChoices)

How a workflow lineage is started.

RunStatus

python
class RunStatus(models.TextChoices)

Execution lifecycle for one pinned workflow run.

StepRunStatus

python
class StepRunStatus(models.TextChoices)

Execution lifecycle for one step-run journal row.

Verdict

python
class Verdict(models.TextChoices)

Resolution lifecycle for one awaited decision slot.

WorkflowQuerySet

python
class WorkflowQuerySet(AngeeQuerySet[Any])

QuerySet owning subject declaration discovery and version currency.

current_published

python
def current_published() -> Self

Return 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

python
def for_subject_declaration(subject_declaration: str) -> Self

Return current published workflows accepting subject_declaration.

WorkflowManager

python
class WorkflowManager(AngeeManager.from_queryset(WorkflowQuerySet))

Manager owning workflow lineage lookups.

current_published_for

python
def current_published_for(workflow: Any) -> Any | None

Return the latest published version for workflow's lineage.

Composes the same _CURRENCY_STATUSES rule current_published owns, scoped to one explicit lineage pool.

WorkflowRunQuerySet

python
class WorkflowRunQuerySet(AngeeQuerySet[Any])

QuerySet owning workflow-run subject lookups.

for_subject

python
def for_subject(subject: Any) -> Self

Return runs whose generic subject is subject.

WorkflowRunManager

python
class WorkflowRunManager(AngeeManager.from_queryset(WorkflowRunQuerySet))

Manager owning workflow-run subject lookups.

Workflow

python
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

python
class Meta()

Django model options for workflow definitions.

__str__

python
def __str__() -> str

Return the workflow's display label.

after_resource_load

python
@classmethod
def after_resource_load(cls,
                        instances: Iterable[Any],
                        *,
                        tier: str,
                        source: str,
                        publish: bool = False) -> None

Reconcile stable keys and publish loaded drafts when requested.

mark_published

python
@transition(status,
            source=WorkflowStatus.DRAFT,
            target=WorkflowStatus.PUBLISHED,
            on_success=_save_workflow_status)
def mark_published() -> None

Mark this copied version as published.

archive

python
@transition(
    status,
    source=WorkflowStatus.PUBLISHED,
    target=WorkflowStatus.ARCHIVED,
    on_success=_save_workflow_status,
)
def archive() -> None

Archive a published workflow version.

clean

python
def clean() -> None

Validate lineage-owned links and normalize stable and subject keys.

validate_subject_declaration

python
def validate_subject_declaration(subject: Any) -> None

Raise when subject does not satisfy this workflow's subject declaration.

save

python
def save(*args: Any, **kwargs: Any) -> None

Persist the workflow after enforcing immutability and model validation.

delete

python
def delete(*args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]

Delete only mutable workflow rows.

publish

python
def publish() -> Self

Copy this draft lineage head into an immutable published version.

publish_if_changed

python
def publish_if_changed() -> Self | None

Publish this draft only when no current version has the same definition.

is_immutable

python
@property
def is_immutable() -> bool

Return whether this workflow version rejects definition edits.

Step

python
class Step(ImplDefaultsMixin, AuditMixin, AngeeDataModel)

One node in a workflow definition graph.

Meta

python
class Meta()

Django model options for workflow steps.

__str__

python
def __str__() -> str

Return the step's display label.

clean

python
def clean() -> None

Validate the step implementation key and config.

save

python
def save(*args: Any, **kwargs: Any) -> None

Persist the step after enforcing parent immutability and validation.

delete

python
def delete(*args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]

Delete only steps belonging to mutable workflow rows.

Edge

python
class Edge(AuditMixin, AngeeDataModel)

Directed edge between two workflow steps.

Meta

python
class Meta()

Django model options for workflow edges.

__str__

python
def __str__() -> str

Return a compact edge label.

clean

python
def clean() -> None

Validate that an edge is fully contained in one workflow.

save

python
def save(*args: Any, **kwargs: Any) -> None

Persist the edge after enforcing parent immutability and validation.

delete

python
def delete(*args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]

Delete only edges belonging to mutable workflow rows.

TriggerManager

python
class TriggerManager(AngeeManager)

Manager owning trigger row claims and due schedule priming.

claim_due_event

python
def claim_due_event(trigger_id: int, *, timestamp: datetime) -> Any | None

Lock and record one enabled event trigger fire if rate limits allow it.

claim_due_schedule

python
def claim_due_schedule(trigger_id: int, *,
                       timestamp: datetime) -> tuple[Any, datetime] | None

Lock and advance one due schedule trigger if rate limits allow it.

prime_due_schedules

python
def prime_due_schedules(*, timestamp: datetime) -> int

Persist initial fire times for enabled schedules missing next_fire_at.

check_event_trigger_change_publishers

python
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

python
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

python
class Meta()

Django model options for workflow triggers.

__str__

python
def __str__() -> str

Return the trigger's display label.

clean

python
def clean() -> None

Validate lineage ownership and trigger declaration shape.

save

python
def save(*args: Any, **kwargs: Any) -> None

Persist the trigger after model validation.

enable

python
def enable() -> None

Enable this trigger through the model owner.

disable

python
def disable() -> None

Disable this trigger through the model owner.

rate_limit_allows

python
def rate_limit_allows(*, timestamp: datetime) -> bool

Return whether this trigger can fire at timestamp.

record_fire

python
def record_fire(
    *, timestamp: datetime, extra_update_fields: Iterable[str] = ()) -> None

Record one trigger fire and persist rate-limit counters.

condition_matches

python
def condition_matches(sender: type[models.Model],
                      instance: models.Model) -> bool

Return whether this event trigger matches a saved model instance.

initial_fire_at

python
def initial_fire_at(*, now: datetime) -> datetime | None

Return the first persisted due timestamp for this schedule trigger.

compute_next_fire_at

python
def compute_next_fire_at(*, after: datetime, now: datetime) -> datetime | None

Return the next scheduled occurrence after after and later than now.

config_mapping

python
@property
def config_mapping() -> Mapping[str, Any]

Return trigger config when it is a JSON object.

WorkflowRun

python
class WorkflowRun(AuditMixin, RecordRefMixin, AngeeDataModel)

One execution of a pinned published workflow version.

Meta

python
class Meta()

Django model options for workflow runs.

is_terminal

python
@property
def is_terminal() -> bool

Return whether this run has reached a terminal status.

awaiting_decision

python
def awaiting_decision() -> bool

Return whether this run has an unresolved workflow decision.

mark_running

python
@transition(status,
            source=RunStatus.PENDING,
            target=RunStatus.RUNNING,
            on_success=save_state)
def mark_running() -> None

Mark a pending run as actively orchestrating.

resume

python
@transition(status,
            source=RunStatus.WAITING,
            target=RunStatus.RUNNING,
            on_success=save_state)
def resume() -> None

Mark a waiting run as actively orchestrating again.

mark_waiting

python
@transition(
    status,
    source=RunStatus.RUNNING,
    target=RunStatus.WAITING,
    on_success=save_state,
)
def mark_waiting(*, wake_at: Any = None) -> None

Mark a run as waiting on durable external or timer state.

mark_succeeded

python
@transition(
    status,
    source=[RunStatus.RUNNING, RunStatus.WAITING],
    target=RunStatus.SUCCEEDED,
    on_success=save_state,
)
def mark_succeeded() -> None

Mark a run as successful.

mark_failed

python
@transition(
    status,
    source=[RunStatus.PENDING, RunStatus.RUNNING, RunStatus.WAITING],
    target=RunStatus.FAILED,
    on_success=save_state,
)
def mark_failed(error: str = "") -> None

Mark a run as failed with an optional durable error message.

mark_canceled

python
@transition(
    status,
    source=[RunStatus.PENDING, RunStatus.RUNNING, RunStatus.WAITING],
    target=RunStatus.CANCELED,
    on_success=save_state,
)
def mark_canceled() -> None

Mark a run as canceled.

save

python
def save(*args: Any, **kwargs: Any) -> None

Persist the run while keeping trigger dedup keys immutable.

from_db

python
@classmethod
def from_db(cls, db: str | None, field_names: list[str],
            values: list[Any]) -> Self

Capture immutable loaded facts without a save-time SELECT.

debit_budget

python
def debit_budget(delta: Mapping[str, int]) -> None

Atomically add usage deltas to this run's budget ledger.

StepRun

python
class StepRun(AuditMixin, AngeeDataModel)

Journal row for one workflow step execution or system-injected event.

Meta

python
class Meta()

Django model options for workflow step-run journal rows.

is_terminal

python
@property
def is_terminal() -> bool

Return whether this journal row has reached a terminal status.

mark_started

python
@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) -> None

Claim this row for execution.

record_attempt

python
def record_attempt(*, heartbeat_at: Any = None) -> None

Record one implementation invocation for this started row.

mark_waiting

python
@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) -> None

Persist durable wait conditions for this row.

wake

python
def wake(*, at: datetime) -> None

Make this waiting journal row due without changing its state.

Event delivery changes only the durable due time. The engine owns the later WAITINGSTARTED claim and therefore remains the sole scheduler of implementation work.

mark_succeeded

python
@transition(
    status,
    source=[StepRunStatus.STARTED, StepRunStatus.WAITING],
    target=StepRunStatus.SUCCEEDED,
    on_success=save_state,
)
def mark_succeeded(*, output: Any = None, outcome: str = "") -> None

Persist a successful step result.

mark_failed

python
@transition(
    status,
    source=[StepRunStatus.STARTED, StepRunStatus.WAITING],
    target=StepRunStatus.FAILED,
    on_success=save_state,
)
def mark_failed(*,
                error: str = "",
                stacktrace: str = "",
                outcome: str = "failed") -> None

Persist a failed step result.

mark_skipped

python
@transition(
    status,
    source=[StepRunStatus.SCHEDULED, StepRunStatus.WAITING],
    target=StepRunStatus.SKIPPED,
    on_success=save_state,
)
def mark_skipped() -> None

Mark this row as skipped by routing or join semantics.

mark_canceled

python
@transition(
    status,
    source=[
        StepRunStatus.SCHEDULED, StepRunStatus.STARTED, StepRunStatus.WAITING
    ],
    target=StepRunStatus.CANCELED,
    on_success=save_state,
)
def mark_canceled() -> None

Mark this row as canceled.

reschedule_for_override

python
@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) -> None

Reset a terminal journal row so a manual override can run it again.

Decision

python
class Decision(AuditMixin, AngeeDataModel)

One awaited resolution slot for a suspended step-run.

Meta

python
class Meta()

Django model options for workflow decisions.

is_terminal

python
@property
def is_terminal() -> bool

Return whether this decision has a terminal verdict.

form_schema_annotation

python
@classmethod
def form_schema_annotation(cls) -> dict[str, Any]

Return the narrow ORM projection consumed by :attr:form_schema.

form_schema

python
@property
def form_schema() -> dict[str, Any] | None

Return the enforced JSON-authored form schema, excluding Python model schemas.

mark_completed

python
@transition(verdict,
            source=Verdict.PENDING,
            target=Verdict.COMPLETED,
            on_success=save_state)
def mark_completed(*, resolution: Any = None, resolved_by: str = "") -> None

Resolve this slot as completed.

mark_rejected

python
@transition(verdict,
            source=Verdict.PENDING,
            target=Verdict.REJECTED,
            on_success=save_state)
def mark_rejected(*, resolution: Any = None, resolved_by: str = "") -> None

Resolve this slot as rejected.

mark_escalated

python
@transition(verdict,
            source=Verdict.PENDING,
            target=Verdict.ESCALATED,
            on_success=save_state)
def mark_escalated(*, resolution: Any = None, resolved_by: str = "") -> None

Resolve this slot as escalated.

mark_expired

python
@transition(verdict,
            source=Verdict.PENDING,
            target=Verdict.EXPIRED,
            on_success=save_state)
def mark_expired(*, resolution: Any = None, resolved_by: str = "") -> None

Resolve this slot as expired.

record_invalid_resolution

python
def record_invalid_resolution() -> None

Record one failed validation attempt while leaving the slot pending.

resolve

python
def resolve(verdict: Verdict,
            *,
            resolution: Any = None,
            resolved_by: str = "") -> None

Resolve this slot through the transition matching verdict.

Released under the AGPL-3.0 License.