alc.models

The domain model layer — the public API is the package __init__. See Architecture Overview for layering rules.

Task

alc.models.task

TaskManifest

TaskManifest(
    input_val: Union[
        str, Path, PathObject, Dict[str, Any], None
    ] = None,
    **kwargs
)

Bases: BaseModel

Standard Alchemy 'Smart' Constructor.

Source code in alc\models\task.py

def __init__(
    self,
    input_val: Union[str, Path, PathObject, Dict[str, Any], None] = None,
    **kwargs,
):
    """Standard Alchemy 'Smart' Constructor."""
    if input_val is not None:
        if isinstance(input_val, dict):
            kwargs.update(input_val)
        elif isinstance(input_val, PathObject):
            kwargs["path_object"] = input_val
        else:
            # Handle raw file string or Path object
            kwargs["input_path"] = input_val
    super().__init__(**kwargs)

task_id property

task_id: str

The manufacturing UID used for filenames.

contributors property

contributors: set

Logins that have actually done work on this task — anyone with a version in the tree. The publish pseudo-user (the publish snapshot) is not a person, so it's excluded. This is the "you have a path with your username in it" signal that predates explicit assignment.

path property

path: Path

The physical location on disk for this .models file.

task_thumb property

task_thumb: dict

The task-level thumbnail resolved live from this task's address (the published render-tree .thumb/thumb.png) → {exists, path} with a root-relative path for /alcroot.

Mirrors the registry entity's asset_thumb: address-derived, so it renders even though the stored thumbnail MediaReference was never populated (its relative_path= constructor arg is a computed field and is silently dropped, leaving path null on disk). This is the reliable path — templates should prefer it over thumbnail.path.

load classmethod

load(path: Union[str, Path]) -> Optional[TaskManifest]

Explicitly load a manifest from a file path. Returns None if missing.

Source code in alc\models\task.py

@classmethod
def load(cls, path: Union[str, Path]) -> Optional[TaskManifest]:
    """Explicitly load a manifest from a file path. Returns None if missing."""
    target = Path(path)
    if not target.exists():
        logger.warning(f"Manifest path missing on disk: {target}")
        return None
    return cls(input_path=target)

hydrate_task_msd classmethod

hydrate_task_msd(data: Any) -> Any

The Universal Entry Point: Handles strings, PathObjects, raw dicts, and parameter addressing combinations. Resolves files seamlessly.

Source code in alc\models\task.py

@model_validator(mode="before")
@classmethod
def hydrate_task_msd(cls, data: Any) -> Any:
    """
    The Universal Entry Point: Handles strings, PathObjects, raw dicts,
    and parameter addressing combinations. Resolves files seamlessly.
    """

    if not isinstance(data, dict):
        return data

    raw_data = {}
    input_path = data.get("input_path", None)
    po = data.get("path_object", None)

    # 1. Branch Extraction Logic
    if input_path:
        raw_data = load_json(input_path)
        # The file's own directory IS its task-msd root — pin it so later
        # commits land back where the manifest was read from, independent
        # of the globally-current selection.
        raw_data["root"] = Path(input_path).parent.as_posix()
    elif po:
        # Rehydrate from disk if this task already exists; only fall back to
        # a fresh baseline for a brand-new task. Without this, building a
        # TaskManifest from a PathObject (e.g. create.task()) and committing
        # would clobber a started task's status / user_versions / sg_id.
        parts = [
            po.project,
            po.episode,
            po.entity_type,
            po.parent,
            po.group,
            po.child,
            po.task,
        ]
        existing_task_id = "_".join([str(p) for p in parts if p]).lower()
        existing_path = Path(po.task_msd_root) / f"{existing_task_id}.json"
        if existing_path.exists():
            raw_data = load_json(existing_path)
            raw_data["root"] = Path(po.task_msd_root).as_posix()
        else:
            raw_data = dict_from_path_object(po)
    else:
        # Look up addressing combinations directly from parameters
        project = data.get("project")
        entity_type = data.get("entity_type")
        parent = data.get("parent")
        child = data.get("child")
        task = data.get("task")

        # Check if all fields required to calculate a disk path are present
        if all([project, entity_type, parent, child, task]):
            # Explicit root wins; otherwise ground to the ACTIVE project
            # (request-scoped in the web app, CFG-current elsewhere).
            if data.get("root"):
                root_dir = Path(data["root"])
            else:
                from alc.context import active_project

                root_dir = Path(active_project().task_msd_root)
                data["root"] = root_dir.as_posix()

            # Derive UID parts
            parts = [
                project,
                entity_type,
                parent,
                child,
                task,
            ]
            task_id = "_".join([str(p) for p in parts if p]).lower()

            # Build the implied JSON file path
            # Match this logic exactly to your actual project storage directory structure
            computed_path = root_dir / f"{task_id}.json"

            if computed_path.exists():
                raw_data = load_json(computed_path)
            else:
                raw_data = data  # Serve baseline defaults safely

    # Merge parameter definitions on top of loaded file data
    # Ensures that arguments like custom root definitions aren't overwritten
    merged_data = {
        **raw_data,
        **{k: v for k, v in data.items() if k not in ["input_path", "path_object"]},
    }

    # 2. Path Normalization
    root = merged_data.get("root", "")
    if root and "/entries" in root:
        merged_data["root"] = root.replace("/entries", "")

    # 3. UserVersions Migration/Hydration
    if "user_versions" in merged_data:
        uv_map = {}
        for user, uv_data in merged_data["user_versions"].items():
            raw_v_data = uv_data.get("version_map", uv_data.get("versions", {}))
            if isinstance(raw_v_data, list):
                raw_v_data = {v["version"]: v for v in raw_v_data}
            uv_map[user] = {"user": user, "version_map": raw_v_data}
        merged_data["user_versions"] = uv_map

    return merged_data

validate_task_logic

validate_task_logic() -> TaskManifest

Post-initialization adjustments.

Source code in alc\models\task.py

@model_validator(mode="after")
def validate_task_logic(self) -> TaskManifest:
    """Post-initialization adjustments."""
    if self.entity_type == "sequences":
        self.entity_type = "shots"

    if self.est_in_days and not self.est_in_hours:
        self.est_in_hours = int(self.est_in_days * 8)
    if self.est_in_hours and not self.est_in_mins:
        self.est_in_mins = int(self.est_in_hours * 60)

    return self

has_versions_for

has_versions_for(user: str) -> bool

Does user have a version rail on this task?

The question every "start the task if it hasn't been started" guard actually means. Those guards used to ask if not task_msd.user_versions — whether the register is empty at all — which is a different question and answers wrong for the one case it matters: a task that carries the publish pseudo-user and nobody else. That happens whenever a publish arrives without a local rail behind it (a migrated project, a hand-placed publish, work published from another machine): the register reads non-empty, the guard skips, and the artist lands on a version view with a publish chip and an empty VERSIONS rail — prp/axe/mdl in loj.

The publish pseudo-user is not a person and has no VERSIONS rail, so it can never satisfy this for anyone.

Source code in alc\models\task.py

def has_versions_for(self, user: str) -> bool:
    """Does ``user`` have a version rail on this task?

    The question every "start the task if it hasn't been started" guard
    actually means. Those guards used to ask ``if not task_msd.user_versions``
    — whether the register is empty *at all* — which is a different question
    and answers wrong for the one case it matters: a task that carries the
    ``publish`` pseudo-user and nobody else. That happens whenever a publish
    arrives without a local rail behind it (a migrated project, a hand-placed
    publish, work published from another machine): the register reads
    non-empty, the guard skips, and the artist lands on a version view with a
    publish chip and an empty VERSIONS rail — ``prp/axe/mdl`` in ``loj``.

    The ``publish`` pseudo-user is not a person and has no VERSIONS rail, so
    it can never satisfy this for anyone.
    """
    if not user or user.lower() == PUBLISH_USER:
        return False
    group = (self.user_versions or {}).get(user)
    return bool(group and group.version_map)

is_assigned_to

is_assigned_to(keys) -> bool

Is this task "mine" for a set of assignment keys? True when I'm an explicit assignee OR I've contributed a version. Keys come from alc.services.people.assignment_keys (email + aliases + local login), so both the canonical email and legacy login strings match.

Source code in alc\models\task.py

def is_assigned_to(self, keys) -> bool:
    """Is this task "mine" for a set of assignment keys? True when I'm an
    explicit ``assignee`` OR I've contributed a version. Keys come from
    ``alc.services.people.assignment_keys`` (email + aliases + local
    login), so both the canonical email and legacy login strings match."""
    wanted = {str(k).strip().lower() for k in keys if k}
    if not wanted:
        return False
    mine = {str(a).strip().lower() for a in (self.assignees or [])}
    mine |= {c.lower() for c in self.contributors}
    return bool(wanted & mine)

commit

commit()

Atomic write to disk.

Source code in alc\models\task.py

def commit(self):
    """Atomic write to disk."""
    from alc.path.io import save_config_data

    save_config_data(self.path, self.model_dump(mode="json"))
    # Keep the master asset registry in lockstep with the task on every save.
    self._ensure_asset_registered()
    # ...and the in-memory task registry, which is hydrated once per project
    # and read by My Tasks / the tracker. Without this a commit is invisible
    # to every view until the process restarts.
    self._sync_registry()
    print(f"Committed MSD: {self.path}")
    logger.info(f"Committed MSD: {self.path}")

update_user_versions

update_user_versions(
    po: PathObject,
    status: Optional[str] = None,
    comment: Optional[str] = None,
)

Register po's version on this task's rail.

comment is the artist's statement about the version (#3796). It is passed explicitly because a PathObject has no comment field — the old getattr(po, "comment", "") could only ever produce "", so every pointer in every register carried a blank comment and the version story lost the one thing a person actually said. An explicit None still falls back to the PathObject, preserving old callers.

Source code in alc\models\task.py

def update_user_versions(
    self,
    po: PathObject,
    status: Optional[str] = None,
    comment: Optional[str] = None,
):
    """Register ``po``'s version on this task's rail.

    ``comment`` is the artist's statement about the version (#3796). It is
    passed explicitly because a ``PathObject`` has no ``comment`` field —
    the old ``getattr(po, "comment", "")`` could only ever produce ``""``,
    so every pointer in every register carried a blank comment and the
    version story lost the one thing a person actually said. An explicit
    ``None`` still falls back to the PathObject, preserving old callers.
    """
    user_name = po.user
    version_str = po.version

    if user_name not in self.user_versions:
        self.user_versions[user_name] = UserVersions(user=user_name)
    print(1)
    user_entry = self.user_versions[user_name]
    version_thumb = po.get_thumb_path(
        thumb_type="version", relative=True, as_posix=True
    )
    version_preview = po.get_preview_path(relative=True, as_posix=True)
    msd_path = remove_root(po.msd_path, as_posix=True)
    print(2)
    vp = VersionPointer(
        version=version_str,
        msd_path=msd_path,
        user=user_name,
        timestamp=datetime.utcnow().isoformat() + "Z",
        comment=(comment if comment is not None else getattr(po, "comment", "")),
        status=status or po.status,
        thumbnail=MediaReference(relative_path=version_thumb),
        preview=MediaReference(relative_path=version_preview),
    )

    user_entry.version_map[version_str] = vp
    # Starting work assigns you: record the author as an explicit assignee
    # (the publish pseudo-user excepted) so the assignment is visible to
    # producers, the tracker, CSV export, and SG push — not only derivable
    # from the version tree. Matches on login; My Tasks resolves it via the
    # people registry (email/alias/local-login).
    if (
        user_name
        and user_name.lower() != PUBLISH_USER
        and user_name not in self.assignees
    ):
        self.assignees.append(user_name)
    if status:
        self.status = status

    self.modified = datetime.utcnow().isoformat() + "Z"
    self.commit()
    logger.info(f"Task {self.task_id} updated: {user_name} created {version_str}")

push

push(status: str = None)

Ensure the task exists in the active tracker and set its status.

Source code in alc\models\task.py

def push(self, status: str = None):
    """Ensure the task exists in the active tracker and set its status."""
    po = self._tracker_po()
    if not po:
        logger.warning("[TaskManifest.push] incomplete address; cannot sync")
        return None
    from alc.tracking import get_tracker

    tracker = get_tracker()
    tracker.ensure_task(po)
    tracked = tracker.set_task_status(po, status or self.status)
    if tracked:
        self.sg_id = tracked.native_id
        self.commit()
    return tracked

pull

pull()

Read the task's remote state into this manifest.

Source code in alc\models\task.py

def pull(self):
    """Read the task's remote state into this manifest."""
    po = self._tracker_po()
    if not po:
        return None
    from alc.tracking import get_tracker

    tracked = get_tracker().get_task(po)
    if tracked:
        self.sg_id = tracked.native_id
        if tracked.status:
            self.status = tracked.status
        self.commit()
    return tracked

Version

alc.models.version

VersionStatus

Bases: BaseModel

Tracks the lifecycle status of a specific version (e.g., 'published', 'approved').

ExportInfo

Bases: BaseModel

The 'Passport' of the MSD: Where did this come from?

VersionManifestData

Bases: BaseModel

Common foundation for all Alchemy Task Manifests.

VersionPointer

Bases: BaseModel

The 'Pointer' artifact used in Task Rail UIs.

UserVersions

Bases: BaseModel

Container for a single user's version history.

rail property

rail: List[VersionPointer]

Returns versions sorted for the Rail UI (001 -> 009).

latest property

latest: Optional[VersionPointer]

Returns the most recent work for this user.

AssetEntry

Bases: BaseModel

Leaf-level asset reference for assembly (Layout/Comp).

StepState

Bases: BaseModel

Tracks the execution state of an individual step.

RecipeRecord

Bases: BaseModel

A record of a specific recipe run (e.g., 'publish').

RecipeWorkManifest

Bases: BaseModel

The master container for all work done on this version.

VersionManifest

Bases: BaseModel

major_minor property

major_minor: tuple[str, str]

Deterministic version splitting.

source_files property

source_files: Dict[str, List[str]]

Dynamic lookup: Always reflects disk state, never stale in the JSON.

render_files property

render_files: Dict[str, List[str]]

Dynamic lookup: Always reflects disk state.

load_msd classmethod

load_msd(
    input_val: Union[str, Path, PathObject],
) -> "VersionManifest"

Hydrates from disk or falls back to PathObject parsing.

Source code in alc\models\version.py

@classmethod
def load_msd(cls, input_val: Union[str, Path, PathObject]) -> "VersionManifest":
    """Hydrates from disk or falls back to PathObject parsing."""
    if isinstance(input_val, PathObject):
        return cls.from_po(input_val)
    target_path = Path(input_val)
    msd_path = (
        target_path
        if target_path.suffix == ".msd"
        else target_path.with_suffix(".msd")
    )

    # Just wrap the PathObject logic
    po = PathObject(str(msd_path))
    return cls.from_po(po)

from_po classmethod

from_po(po: 'PathObject') -> 'VersionManifest'

The Single Source of Truth: Merges disk data (if any) with PathObject attrs.

Source code in alc\models\version.py

@classmethod
def from_po(cls, po: "PathObject") -> "VersionManifest":
    """
    The Single Source of Truth: Merges disk data (if any) with PathObject attrs.
    """
    # 1. Start with data from disk if it exists
    disk_data = {}
    msd_path = po.msd_path
    if msd_path.exists():
        disk_data = load_json(msd_path)

    # 2. Extract data from the PathObject
    po_data = po.model_dump() if hasattr(po, "model_dump") else po.__dict__

    # 3. Merge: PathObject values override disk values
    # We only take keys that actually exist in the VersionManifest model
    manifest_fields = cls.model_fields.keys()
    for key, value in po_data.items():
        if key in manifest_fields and value is not None:
            disk_data[key] = value

    # 4. Finalize specific references
    disk_data["path_object"] = PathObject(msd_path)

    # 5. One single instantiation
    return cls(**disk_data)

sync_path_object_data

sync_path_object_data() -> VersionManifest

Hydrates manifest fields from the PathObject context. Replaces legacy post_init.

Source code in alc\models\version.py

@model_validator(mode="after")
def sync_path_object_data(self) -> VersionManifest:
    """
    Hydrates manifest fields from the PathObject context.
    Replaces legacy __post_init__.
    """
    if self.path_object:
        self.task_id = self.path_object.task_id
        # Auto-calculate preview/thumb paths if not explicitly set
        if not self.preview.relative_path:
            self.preview.path = self.path_object.get_preview_path(relative=True)
        if not self.thumbnail.relative_path:
            self.thumbnail.path = self.path_object.get_thumb_path(
                thumb_type="version", relative=True
            )
    return self

update_recipe_step

update_recipe_step(
    recipe_type: str,
    step_name: str,
    status: StepStatus,
    message: str = "",
    last_run: str = "",
    shared_data: dict = None,
)

High-level helper to record progress of a recipe step.

Parameters:

Name	Type	Description	Default
recipe_type	str	e.g., 'publish', 'render', 'build'	required
step_name	str	The unique name of the step from the dsn.json	required
status	StepStatus	A StepStatus enum value (passed, failed, running, etc.)	required
message	str	Optional detail (e.g., error logs or success notes)	''
shared_data	dict	Optional dictionary to merge into the recipe's shared_data	None

Source code in alc\models\version.py

def update_recipe_step(
    self,
    recipe_type: str,
    step_name: str,
    status: StepStatus,
    message: str = "",
    last_run: str = "",
    shared_data: dict = None,
):
    """
    High-level helper to record progress of a recipe step.

    Args:
        recipe_type: e.g., 'publish', 'render', 'build'
        step_name: The unique name of the step from the dsn.json
        status: A StepStatus enum value (passed, failed, running, etc.)
        message: Optional detail (e.g., error logs or success notes)
        shared_data: Optional dictionary to merge into the recipe's shared_data
    """
    # 1. Get or create the specific recipe record (e.g., the 'publish' bucket)
    record = self.work_log.get_record(recipe_type)

    # 2. Update the step status and message
    record.update_step(step_name, status, msg=message)

    # 3. If shared data was provided, merge it into the recipe record
    if shared_data:
        record.shared_data.update(shared_data)

    # 4. Commit to disk to ensure the MSD stays in sync
    self.commit()

update_status

update_status(
    new_status: str,
    comment="Automated status update by Alchemy",
    user: Optional[str] = None,
    origin: str = "system",
)

Convenience method to update status and commit.

Source code in alc\models\version.py

def update_status(
    self,
    new_status: str,
    comment="Automated status update by Alchemy",
    user: Optional[str] = None,
    origin: str = "system",
):
    """Convenience method to update status and commit."""
    self.status = new_status
    vstatus = VersionStatus(
        status=self.status,
        user=user or self.path_object.user,
        comment=comment,
        origin=origin,
    )
    self.status_history.append(vstatus)
    self.commit()

push

push(
    status: str = None,
    description: str = None,
    review: bool = False,
)

Create/sync this version in the active tracker, stamping sg_id locally.

Returns the canonical TrackerVersion (or None). review=True also pushes preview media and caches the dailies url.

Source code in alc\models\version.py

def push(self, status: str = None, description: str = None, review: bool = False):
    """Create/sync this version in the active tracker, stamping sg_id locally.

    Returns the canonical ``TrackerVersion`` (or None). ``review=True`` also
    pushes preview media and caches the dailies url.
    """
    if not self.path_object:
        logger.warning("[VersionManifest.push] no path_object; cannot sync")
        return None
    from alc.tracking import get_tracker

    tracker = get_tracker()
    tracked = tracker.create_version(
        self.path_object,
        status=status or self.status,
        description=description or self.description,
    )
    if not tracked:
        return None
    self.sg_id = tracked.native_id
    if review:
        _, dailies_url = tracker.submit_review(self.path_object)
        if dailies_url:
            self.dailies_url = dailies_url
    self.commit()
    return tracked

pull

pull()

Read the version's remote state into this manifest (no history spam).

Source code in alc\models\version.py

def pull(self):
    """Read the version's remote state into this manifest (no history spam)."""
    if not self.path_object:
        return None
    from alc.tracking import get_tracker

    tracked = get_tracker().get_version(self.path_object)
    if tracked:
        self.sg_id = tracked.native_id
        if tracked.status:
            self.status = tracked.status
        self.commit()
    return tracked

commit

commit() -> str | None

The Alchemy Transaction. Uses model_dump(exclude=...) to replace the legacy 'to_dict' sifter.

Source code in alc\models\version.py

def commit(self) -> str | None:
    """
    The Alchemy Transaction.
    Uses model_dump(exclude=...) to replace the legacy 'to_dict' sifter.
    """
    from alc.path import save_json

    print("self.data", self.data)
    # We exclude path_object and volatile props because they are recalculated on load
    payload = self.model_dump(mode="json", exclude={"path_object", "major_minor"})

    # Inject dynamic file lists into the persistent JSON for headless/database use
    payload["source_files"] = self.source_files
    payload["render_files"] = self.render_files

    save_json(self.path, payload)
    print("saving json", self.path)
    logger.info(f"Version MSD Committed: {self.path_object.msd_path}")

    # update the atsk with the new version info as well.
    return self.path

Media

alc.models.media

MediaReference

Bases: BaseModel

Standardized media pointer for Alchemy Artifacts. Provides deterministic path resolution and availability checks.

Contract: path is stored root-relative (relative to the prod root, e.g. nca/tpb/PUBLISHES/render/.../thumb.png) so it is portable across machines and serves directly under the /alcroot mount as /alcroot/{path}. Disk checks (:attr:exists) resolve it to an absolute path against the active prod root on demand — never store an absolute path.

abs_path property

abs_path: Optional[Path]

The stored root-relative path resolved to an absolute path against the active prod root (for disk access). None when unset.

relative_path property

relative_path: str

The root-relative path (what /alcroot serves).

exists property

exists: bool

Active check for file existence. Critical for agents to avoid 'File Not Found' errors during automation. Resolves the root-relative path to absolute before stat-ing.

has_media property

has_media: bool

True if either a local file exists or a remote URL is defined.