Angee

Appearance

angee.work.models ​

Operational queues, cycles, triage, and same-row task work mechanics.

Queue is a materialized child of :class:spaces.Group: its inherited group row remains the one roster and hierarchy owner. TaskWork is a table-less donor onto projects.Task; queue, stage, cycle, numbering, and triage columns therefore exist only when this addon is composed, on the task table itself.

QueueManager ​

python
class QueueManager(GroupManager)

Own personal-queue identity and idempotent provisioning.

personal_slug ​

python
def personal_slug(user: models.Model) -> str

Return the unique stable slug of user's personal queue.

personal_key ​

python
def personal_key(user: models.Model) -> str

Return P<pk>, rejecting PKs above 11 digits rather than truncating.

personal_for ​

python
def personal_for(user: models.Model, *, provision: bool = False) -> Any | None

Return user's personal queue, optionally provisioning it.

provision_personal ​

python
def provision_personal(user: models.Model) -> Any

Idempotently create one private owner-rostered queue for user.

Queue ​

python
class Queue(AngeeModel)

An ongoing operational context, materialized as a spaces.Group child.

EstimateScale ​

python
class EstimateScale(models.TextChoices)

Estimation vocabulary configured per queue.

CycleStartDay ​

python
class CycleStartDay(models.IntegerChoices)

ISO weekday on which generated cycles start.

Meta ​

python
class Meta()

Django model options for the materialized queue child.

clean ​

python
def clean() -> None

Normalize the queue key and validate its explicit default stage.

save ​

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

Persist the queue and atomically provision its stage/numbering substrate.

task_sequence_key ​

python
def task_sequence_key() -> str

Return the stable sequence lookup key for this queue's task numbers.

ensure_task_sequence ​

python
def ensure_task_sequence() -> models.Model

Return this queue's task-number sequence, creating it if absent.

next_task_number ​

python
def next_task_number() -> int

Draw the next gapless task number inside the caller's transaction.

Stage ​

python
class Stage(StagePrimitive, AuditMixin, AngeeDataModel)

One ordered, user-named pipeline stage owned by a queue.

StageCategory ​

python
class StageCategory(models.TextChoices)

Closed semantic categories projected onto coarse task status.

Meta ​

python
class Meta(StagePrimitive.Meta)

Django model options for queue stages.

from_db ​

python
@classmethod
def from_db(cls, db: Any, field_names: Sequence[str],
            values: Sequence[Any]) -> Stage

Load a row and remember facts that identify a system stage.

save ​

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

Prevent user creation or renaming of triage/duplicate system stages.

delete ​

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

Prevent users from deleting the two system-provisioned stages.

CycleManager ​

python
class CycleManager(AngeeManager)

Own idempotent queue-cadence generation and due-cycle closure.

generate_for_queue ​

python
def generate_for_queue(
        queue: models.Model,
        *,
        as_of: date | None = None,
        window_end: date | None = None,
        completed_at: datetime | None = None) -> tuple[Any, ...]

Generate and close one queue's cadence, returning its ordered cycles.

Cycle windows are closed date intervals: both starts_on and ends_on belong to the cycle. The first empty-cadence window starts on the configured weekday at or before as_of (an exact start-day hit is included). Each next start is the day after the inclusive end plus exactly cycle_cooldown_weeks * 7 calendar dates, which belong to no cycle. A cycle closes only when as_of > ends_on; its end day remains active.

Passing window_end generates every cadence start through that inclusive boundary. Without it, generation maintains the queue's configured number of future cycles. The queue row is locked, existing windows are reused, and numbers advance from the last row, making reruns over the same window exact no-ops.

Cycle ​

python
class Cycle(AuditMixin, AngeeDataModel)

One generated, queue-scoped planning window with immutable close evidence.

Meta ​

python
class Meta()

Django model options for generated queue cycles.

display_name ​

python
@property
def display_name() -> str

Return the optional custom name or its stable Cycle N fallback.

__str__ ​

python
def __str__() -> str

Return the custom or generated cycle label.

clean ​

python
def clean() -> None

Reject an inverted generated date window.

save ​

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

Persist while keeping close timestamp and snapshot immutable.

close ​

python
def close(*,
          next_cycle: Any | None = None,
          completed_at: datetime | None = None) -> Cycle

Snapshot open tasks and roll them into the next cycle atomically.

uncompleted_upon_close stores the sorted public task ids exactly as they stood immediately before rollover. Reinvoking a completed cycle is an exact no-op and cannot replace that evidence or move tasks again.

TaskWork ​

python
class TaskWork(StagedModelMixin, AngeeModel)

Same-row work contribution folded into projects.Task.

Meta ​

python
class Meta()

Abstract donor options folded into the concrete task table.

__init__ ​

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

Track caller-authored status separately from Django row loading.

__setattr__ ​

python
def __setattr__(name: str, value: Any) -> None

Remember direct status assignment while allowing the projection writer.

refresh_from_db ​

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

Refresh without misclassifying Django's field hydration as a direct write.

full_clean ​

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

Let Django normalize fields without inventing a direct status write.

Model.clean_fields() assigns every cleaned value back through the field descriptor. Preserve a status assignment already made by the caller, but do not treat that framework-owned normalization as a new authored assignment. This keeps resource loading compatible with the direct-status rejection enforced by :meth:clean and :meth:save.

apply_create_defaults ​

python
def apply_create_defaults() -> Mapping[str, Sequence[Any]]

Default queue/stage from the actor and return their create relations.

clean ​

python
def clean() -> None

Project a stage before base lifecycle validation and enforce scope.

save ​

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

Persist stage projection and queue numbering in one transaction.

complete ​

python
def complete() -> Any

Move to the first completed stage, or use the base verb without a queue.

drop ​

python
def drop(reason: Any) -> Any

Move to duplicate/canceled according to the base dropped reason.

reopen ​

python
def reopen() -> Any

Move to the queue-owned default stage, or use the base verb without a queue.

accept ​

python
def accept(stage: models.Model | None = None) -> Any

Leave triage for a same-queue, non-system stage, idempotently.

decline ​

python
def decline(reason: Any) -> Any

Leave triage for the canceled stage with a closed dropped reason.

snooze ​

python
def snooze(until: datetime) -> Any

Snooze a triage task until an inclusive instant or new chatter activity.

wake_due_snoozes ​

python
@classmethod
def wake_due_snoozes(cls, *, now: datetime | None = None) -> int

Clear snoozes whose inclusive wake instant is at or before now.

wake_from_chatter_thread ​

python
@classmethod
def wake_from_chatter_thread(cls, thread_id: Any) -> int

Clear snooze state on the task whose chatter owns thread_id.

mark_duplicate ​

python
def mark_duplicate(canonical: models.Model) -> Any

Merge this duplicate into canonical in one row-locked transaction.

The atomic postcondition is one directional duplicate relation, this task in its system duplicate stage (and therefore dropped/duplicate), all source links and followers moved with collision deduplication, followed by every deterministic ANGEE_WORK_MERGE_CONTRIBUTORS mover. Any failure rolls every one of those writes back.

work_key ​

python
@property
def work_key() -> str | None

Return the queue-keyed task number (for example ENG-42).

validate_cycle_scope ​

python
def validate_cycle_scope() -> None

Reject a cycle outside the task's queue.

UserWork ​

python
class UserWork(AngeeModel)

Provision personal queues after IAM user resources load.

Meta ​

python
class Meta()

Abstract same-row provisioning donor for IAM users.

after_resource_load ​

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

Ensure every loaded user has exactly one personal queue.

Released under the AGPL-3.0 License.

Copyright © 2025–present Angee, Inc.