Angee

Appearance

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 WAITING → STARTED 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.

Copyright © 2025–present Angee, Inc.