version

version

DomoWorkflow_Version dataclass

DomoWorkflow_Version(
    schema_version: int = 2,
    tiles: list[DomoWorkflow_Tile] = list(),
    edges: list[DomoWorkflow_Edge] = list(),
    layouts: list[dict[str, Any]] = list(),
    inputs: dict[str, DomoWorkflow_Parameter] = dict(),
    outputs: dict[str, DomoWorkflow_Parameter] = dict(),
    *,
    raw: dict = dict(),
    model_id: str = "",
    version: str = "",
    owner_id: str | None = None,
    created_by: str | None = None,
    created_dt: datetime | None = None,
    updated_by: str | None = None,
    updated_dt: datetime | None = None,
    active: bool | None = None,
    deployed_on: datetime | None = None,
    saved_json: str | None = None,
    parent: Any | None = None
)

Bases: DomoWorkflow_Design

A workflow version that merges semver metadata with design behavior.

Attributes:

Name	Type	Description
model_id	str	Parent workflow model UUID.
version	str	Semver string, e.g. "1.0.0".
owner_id	str | None	User ID of the version owner.
created_by	str | None	User ID who created this version.
created_dt	datetime | None	Creation timestamp.
updated_by	str | None	User ID who last updated this version.
updated_dt	datetime | None	Last-update timestamp.
parent	Any | None	Optional parent DomoWorkflow reference.
raw	dict	Raw version API response dict.

is_deployed property

is_deployed: bool

Return True when the workflow version appears to be deployed/active.

from_dict classmethod

from_dict(
    obj: dict[str, Any], parent: Any | None = None
) -> DomoWorkflow_Version

Build a version from a workflow version API payload.

Parameters:

Name	Type	Description	Default
obj	dict[str, Any]	Raw version dict (savedJson is parsed when well-formed).	required
parent	Any | None	Optional parent DomoWorkflow.	None

Returns:

Type	Description
DomoWorkflow_Version	The populated version; the design is empty when savedJson is
DomoWorkflow_Version	missing or malformed.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

@classmethod
def from_dict(
    cls,
    obj: dict[str, Any],
    parent: Any | None = None,  # DomoWorkflow — avoid circular import
) -> DomoWorkflow_Version:
    """Build a version from a workflow version API payload.

    Args:
        obj: Raw version dict (``savedJson`` is parsed when well-formed).
        parent: Optional parent ``DomoWorkflow``.

    Returns:
        The populated version; the design is empty when ``savedJson`` is
        missing or malformed.
    """
    saved_json_str = obj.get("savedJson")
    design = DomoWorkflow_Design.empty()
    if saved_json_str:
        # Malformed savedJson — keep definition as None
        with contextlib.suppress(ValueError, KeyError, TypeError, AttributeError):
            design = DomoWorkflow_Design.from_raw(json.loads(saved_json_str))

    return cls(
        schema_version=design.schema_version,
        tiles=design.tiles,
        edges=design.edges,
        layouts=design.layouts,
        inputs=design.inputs,
        outputs=design.outputs,
        model_id=obj.get("modelId", ""),
        version=obj.get("version", ""),
        owner_id=obj.get("owner"),
        created_by=obj.get("createdBy"),
        created_dt=convert_string_to_datetime(obj.get("createdOn")),
        updated_by=obj.get("updatedBy"),
        updated_dt=convert_string_to_datetime(obj.get("updatedOn")),
        active=obj.get("active"),
        deployed_on=convert_string_to_datetime(obj.get("deployedOn")),
        saved_json=saved_json_str,
        parent=parent,
        raw=obj,
    )

get_definition async

get_definition(
    *,
    auth: DomoAuth | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the v2 definition of this workflow version.

Parameters:

Name	Type	Description	Default
auth	DomoAuth | None	Session to use; defaults to the parent workflow's auth.	None
return_raw	bool	If True, return the response without raising on failure.	False
debug_api	bool	If True, enable API debug output.	False
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

Raises:

Type	Description
ValueError	If no auth is given and the version has no parent auth.
Workflow_GET_Error	If the response is not successful and return_raw is False.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

async def get_definition(
    self,
    *,
    auth: DomoAuth | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the v2 definition of this workflow version.

    Args:
        auth: Session to use; defaults to the parent workflow's auth.
        return_raw: If True, return the response without raising on failure.
        debug_api: If True, enable API debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.

    Raises:
        ValueError: If no auth is given and the version has no parent auth.
        Workflow_GET_Error: If the response is not successful and
            ``return_raw`` is False.
    """
    resolved_auth = auth or getattr(self.parent, "auth", None)
    if resolved_auth is None:
        raise ValueError("auth is required when version parent has no auth")

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await workflow_routes.get_workflow_version_definition(
        auth=resolved_auth,
        workflow_id=self.model_id,
        version_number=self.version,
        return_raw=return_raw,
        context=context,
    )

get_trigger_form async

get_trigger_form(
    *,
    auth: DomoAuth | None = None,
    trigger_id: str | None = None,
    deployment_id: str | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the trigger form of this workflow version.

Parameters:

Name	Type	Description	Default
auth	DomoAuth | None	Session to use; defaults to the parent workflow's auth.	None
trigger_id	str | None	Optional trigger to render the form for.	None
deployment_id	str | None	Optional deployment to render the form for.	None
return_raw	bool	If True, return the response without raising on failure.	False
debug_api	bool	If True, enable API debug output.	False
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

Raises:

Type	Description
ValueError	If no auth is given and the version has no parent auth.
Workflow_GET_Error	If the response is not successful and return_raw is False.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

async def get_trigger_form(
    self,
    *,
    auth: DomoAuth | None = None,
    trigger_id: str | None = None,
    deployment_id: str | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the trigger form of this workflow version.

    Args:
        auth: Session to use; defaults to the parent workflow's auth.
        trigger_id: Optional trigger to render the form for.
        deployment_id: Optional deployment to render the form for.
        return_raw: If True, return the response without raising on failure.
        debug_api: If True, enable API debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.

    Raises:
        ValueError: If no auth is given and the version has no parent auth.
        Workflow_GET_Error: If the response is not successful and
            ``return_raw`` is False.
    """
    resolved_auth = auth or getattr(self.parent, "auth", None)
    if resolved_auth is None:
        raise ValueError("auth is required when version parent has no auth")

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await workflow_routes.get_workflow_trigger_form(
        auth=resolved_auth,
        model_id=self.model_id,
        version=self.version,
        trigger_id=trigger_id,
        deployment_id=deployment_id,
        return_raw=return_raw,
        context=context,
    )

release async

release(
    *,
    auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Release (deploy/activate) this specific workflow version.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

async def release(
    self,
    *,
    auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Release (deploy/activate) this specific workflow version."""
    resolved_auth = auth or getattr(self.parent, "auth", None)
    if resolved_auth is None:
        raise ValueError("auth is required when version parent has no auth")

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await workflow_routes.release_workflow_version(
        auth=resolved_auth,
        model_id=self.model_id,
        version=self.version,
        context=context,
    )

to_dict

to_dict() -> dict[str, Any]

Serialize to the API shape (suitable for PUT /versions/{version}).

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to the API shape (suitable for PUT /versions/{version})."""
    payload: dict[str, Any] = {
        "modelId": self.model_id,
        "version": self.version,
        "owner": self.owner_id,
        "createdBy": self.created_by,
        "createdOn": self.created_dt.isoformat() if self.created_dt else None,
        "updatedBy": self.updated_by,
        "updatedOn": self.updated_dt.isoformat() if self.updated_dt else None,
        "active": self.active,
        "deployedOn": self.deployed_on.isoformat() if self.deployed_on else None,
    }
    payload["savedJson"] = json.dumps(super().to_dict())
    return payload

validate_definition

validate_definition(
    *,
    require_true_false_labels: bool = True,
    logic_tile_types: tuple[str, ...] = (
        "conditionalGatewayNode",
    ),
    allow_dangling_tile_types: tuple[str, ...] = (
        "endNode",
    )
) -> tuple[bool, list[str]]

Validate this version's design topology and branching rules.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

def validate_definition(
    self,
    *,
    require_true_false_labels: bool = True,
    logic_tile_types: tuple[str, ...] = ("conditionalGatewayNode",),
    allow_dangling_tile_types: tuple[str, ...] = ("endNode",),
) -> tuple[bool, list[str]]:
    """Validate this version's design topology and branching rules."""
    if not self.tiles:
        return False, [f"missing definition for workflow version '{self.version}'"]

    errors: list[str] = []
    if require_true_false_labels:
        try:
            self.validate_boolean_conditionals(required_labels=("True", "False"))
        except ValueError as exc:
            errors.append(str(exc))

    errors.extend(
        self.get_connection_validation_errors(
            logic_tile_types=logic_tile_types,
            allow_dangling_tile_types=allow_dangling_tile_types,
            required_logic_outputs=2,
        )
    )
    return len(errors) == 0, errors

validate_with_domo async

validate_with_domo(
    *,
    auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Validate this version's design with Domo's official validator API.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

async def validate_with_domo(
    self,
    *,
    auth: DomoAuth | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Validate this version's design with Domo's official validator API."""
    if not self.tiles:
        raise ValueError(f"missing design for workflow version '{self.version}'")

    resolved_auth = auth or getattr(self.parent, "auth", None)
    if resolved_auth is None:
        raise ValueError("auth is required when version parent has no auth")

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await workflow_routes.validate_workflow_model(
        auth=resolved_auth,
        body=DomoWorkflow_Design.to_dict(self),
        context=context,
    )

DomoWorkflow_Versions dataclass

DomoWorkflow_Versions(
    auth: DomoAuth,
    model_id: str,
    parent: Any | None = None,
    items: list[DomoWorkflow_Version] | None = None,
    current_version: DomoWorkflow_Version | None = None,
)

Bases: DomoManager

Manager for loading workflow versions where each version is a design object.

get async

get(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoWorkflow_Version]

List all versions for a workflow model.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

async def get(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoWorkflow_Version]:
    """List all versions for a workflow model."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await workflow_routes.list_workflow_versions(
        auth=self.auth,
        model_id=self.model_id,
        context=context,
    )

    versions_raw = res.response if res.is_success else []
    self.items = [
        DomoWorkflow_Version.from_dict(v, parent=self.parent)
        for v in (versions_raw or [])
    ]
    return self.items

get_by_version async

get_by_version(
    version: str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow_Version | None

Fetch one specific version for this workflow model.

Source code in src/crew_dcs/classes/DomoWorkflow/version.py

async def get_by_version(
    self,
    version: str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow_Version | None:
    """Fetch one specific version for this workflow model."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await workflow_routes.get_workflow(
        auth=self.auth,
        model_id=self.model_id,
        version_id=version,
        context=context,
    )

    if not res.is_success:
        return None

    self.current_version = DomoWorkflow_Version.from_dict(
        res.response,
        parent=self.parent,
    )
    return self.current_version