Skip to content

Workflows

albert.collections.workflows.WorkflowCollection

WorkflowCollection(*, session: AlbertSession)

Bases: BaseCollection

Manage Workflows in the Albert platform.

A Workflow is a grouping of parameters and their setpoints, the independent variables a test is run under. It is built from one or more groupings, where each grouping is either a Data Template with pre-linked parameters or a Parameter Group. A single Workflow can combine a Data Template's pre-linked parameters with one or more Parameter Groups, each contributing its own parameter setpoints (a value, plus a unit where applicable). Because a Data Template can carry pre-linked parameters, it is used here exactly like a Parameter Group: purely to describe parameters and their setpoints.

A Workflow does not include a Data Template's Data Columns (also called Results). Those are the dependent variables, and they are recorded only in Property Data (PropertyDataCollection). In short, the Workflow holds the independent variables and Property Data holds the dependent ones. A Workflow is also not itself a task: it becomes actionable when paired with a Data Template inside a Block on a Property or Batch Task (see add_block).

Uniqueness. A Workflow is uniquely identified by its full setpoint configuration: the value (and unit) of every parameter setpoint, the order of the parameters within each Data Template / Parameter Group, and the order of the Data Templates / Parameter Groups within the workflow. Because of this, workflows are found-or-created rather than blindly created: to obtain a workflow ID, build the Workflow object you want and let create return the existing match or make a new one.

Every tenant also includes a built-in workflow WFL1 ("No Parameter Group") with no parameter groups. Use it for tasks and blocks that do not involve parameter groups; it does not need to be created via create.

Intervals. When one or two parameters are "intervalized" (varied across several values), the workflow acts as a parent that carries the resulting interval combinations. Each combination has an interval ID of the form ROW1 (one intervalized parameter) or ROW1XROW2 (the product of two). That interval ID is how you target a specific condition when reading or writing Property Data. Use get_interval_id to build the correct interval ID from parameter values.

This collection is accessed as client.workflows.

Example

from albert import Albert
client = Albert()
wf = client.workflows.get_by_id(id="WFL1")
# Build the interval ID for a specific condition, then use it with
# client.property_data to read/write that interval's results.
interval_id = wf.get_interval_id({"Temperature": 25})
interval_id
# 'ROW1'

Parameters:

Name Type Description Default
session AlbertSession

The authenticated Albert session used for API calls.

required

Attributes:

Name Type Description
base_path str

The base API route for workflow requests.

Methods:

Name Description
create

Find-or-create workflows, deduplicating by parameter setpoints.

get_all_facets

Get facet groups (aggregated filter counts) for a workflow query.

get_by_id

Get a single workflow, including its full setpoints and interval combinations.

get_by_ids

Get multiple workflows by their IDs in batches.

get_all

Iterate over all workflows (rarely needed in production).

search

Search for workflows matching the given filters.

Parameters:

Name Type Description Default
session AlbertSession

The authenticated Albert session used for API calls.

required
Source code in src/albert/collections/workflows.py
def __init__(self, *, session: AlbertSession):
    """Initialize a WorkflowCollection.

    Parameters
    ----------
    session : AlbertSession
        The authenticated Albert session used for API calls.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{WorkflowCollection._api_version}/workflows"

base_path

base_path = (
    f"/api/{WorkflowCollection._api_version}/workflows"
)

create

create(*, workflows: list[Workflow]) -> list[Workflow]

Find or create workflows.

This is the intended way to obtain a workflow ID: build the Workflow object you want and let this method return the existing match or create a new one. Workflows are deduplicated by their full setpoint configuration: the value and unit of every setpoint, the order of parameters within each Data Template / Parameter Group, and the order of those groups within the workflow. Any parameter group supplied by its ID only is expanded to its full parameters before matching.

Example

from albert.resources.workflows import (
    Workflow,
    ParameterGroupSetpoints,
    ParameterSetpoint,
)

# A workflow combining a Data Template's pre-linked parameters (keyed by a
# DAT... id, used just like a Parameter Group) with two Parameter Groups.
workflow = Workflow(
    parameter_group_setpoints=[
        ParameterGroupSetpoints(
            id="DAT9999999",
            parameter_setpoints=[
                ParameterSetpoint(parameter_id="PRM9999999", value="23"),
                ParameterSetpoint(parameter_id="PRM2", value="50"),
            ],
        ),
        ParameterGroupSetpoints(
            id="PRG9999999",
            parameter_setpoints=[
                ParameterSetpoint(parameter_id="PRM3", value="24"),
            ],
        ),
        ParameterGroupSetpoints(
            id="PRG2",
            parameter_setpoints=[
                ParameterSetpoint(parameter_id="PRM4", value="2000"),
            ],
        ),
    ],
)
created = client.workflows.create(workflows=[workflow])
created[0].id
# 'WFL1'

Parameters:

Name Type Description Default
workflows list[Workflow]

The workflows to find or create. Each is built from parameter group setpoints (see Workflow).

required

Returns:

Type Description
list[Workflow]

The created or matched workflows, in the same order as the input.

Raises:

Type Description
AlbertPartialError

If only some of the workflows could be created or matched. The error carries the succeeded (created_items) and failed (failed_items) items.

Notes

When a group is identified by id, the SDK resolves setpoint sequence row ids from that Data Template or Parameter Group before create. Do not supply sequence on ParameterSetpoint objects.

The platform derives each workflow's name from its parameter groups and setpoints, so a caller-supplied name is not preserved. Returned workflows carry the parameter group setpoints they were created or matched with. A matched workflow without parameter groups (such as the built-in WFL1) comes back without those details, so the SDK fetches it by ID before returning it.

Source code in src/albert/collections/workflows.py
def create(self, *, workflows: list[Workflow]) -> list[Workflow]:
    """Find or create workflows.

    This is the intended way to obtain a workflow ID: build the Workflow object
    you want and let this method return the existing match or create a new one.
    Workflows are deduplicated by their full setpoint configuration: the value
    and unit of every setpoint, the order of parameters within each Data
    Template / Parameter Group, and the order of those groups within the
    workflow. Any parameter group supplied by its ID only is expanded to its
    full parameters before matching.

    !!! example
        ```python
        from albert.resources.workflows import (
            Workflow,
            ParameterGroupSetpoints,
            ParameterSetpoint,
        )

        # A workflow combining a Data Template's pre-linked parameters (keyed by a
        # DAT... id, used just like a Parameter Group) with two Parameter Groups.
        workflow = Workflow(
            parameter_group_setpoints=[
                ParameterGroupSetpoints(
                    id="DAT9999999",
                    parameter_setpoints=[
                        ParameterSetpoint(parameter_id="PRM9999999", value="23"),
                        ParameterSetpoint(parameter_id="PRM2", value="50"),
                    ],
                ),
                ParameterGroupSetpoints(
                    id="PRG9999999",
                    parameter_setpoints=[
                        ParameterSetpoint(parameter_id="PRM3", value="24"),
                    ],
                ),
                ParameterGroupSetpoints(
                    id="PRG2",
                    parameter_setpoints=[
                        ParameterSetpoint(parameter_id="PRM4", value="2000"),
                    ],
                ),
            ],
        )
        created = client.workflows.create(workflows=[workflow])
        created[0].id
        # 'WFL1'
        ```

    Parameters
    ----------
    workflows : list[Workflow]
        The workflows to find or create. Each is built from parameter group
        setpoints (see [`Workflow`][albert.resources.workflows.Workflow]).

    Returns
    -------
    list[Workflow]
        The created or matched workflows, in the same order as the input.

    Raises
    ------
    AlbertPartialError
        If only some of the workflows could be created or matched. The error
        carries the succeeded (``created_items``) and failed (``failed_items``)
        items.

    Notes
    -----
    When a group is identified by ``id``, the SDK resolves setpoint ``sequence`` row
    ids from that Data Template or Parameter Group before create. Do not supply
    ``sequence`` on [`ParameterSetpoint`][albert.resources.workflows.ParameterSetpoint]
    objects.

    The platform derives each workflow's ``name`` from its parameter groups and
    setpoints, so a caller-supplied ``name`` is not preserved. Returned workflows
    carry the parameter group setpoints they were created or matched with. A matched
    workflow without parameter groups (such as the built-in ``WFL1``) comes back
    without those details, so the SDK fetches it by ID before returning it.
    """
    if isinstance(workflows, Workflow):
        # in case the user forgets this should be a list
        workflows = [workflows]

    # Hydrate any parameter groups provided only by ID with their parameters
    for wf in workflows:
        self._hydrate_parameter_groups(workflow=wf)
        for pg_setpoint in wf.parameter_group_setpoints:
            for setpoint in pg_setpoint.parameter_setpoints:
                # The platform accepts shortName only on Special parameters and
                # rejects it with a 400 anywhere else; never send it otherwise.
                if setpoint.category != ParameterCategory.SPECIAL:
                    setpoint.short_name = None

    response = self.session.post(
        url=f"{self.base_path}/bulk",
        json=[
            x.model_dump(
                mode="json",
                by_alias=True,
                exclude_none=True,
                exclude={"created", "updated"},
            )
            for x in workflows
        ],
    )
    results = []
    for x in unpack_bulk_created_items(response):
        if "name" not in x:
            # The platform omits the name of a matched workflow that has no
            # parameter groups; fetch the full record instead.
            target_id = x.get("existingAlbertId") or x.get("albertId")
            if not target_id:
                raise AlbertException(f"Workflow response item missing ID: {x}")
            results.append(self.get_by_id(id=target_id))
        else:
            results.append(Workflow(**x))
    return results

get_by_id

get_by_id(*, id: WorkflowId) -> Workflow

Get a single workflow by its ID, including its full setpoints and interval combinations.

Includes the fully populated parameter_group_setpoints and any interval combinations.

Example

wf = client.workflows.get_by_id(id="WFL1")
wf.id
# 'WFL1'

Parameters:

Name Type Description Default
id WorkflowId

The workflow ID (format WFL...).

required

Returns:

Type Description
Workflow

The fully populated workflow.

Source code in src/albert/collections/workflows.py
@validate_call
def get_by_id(self, *, id: WorkflowId) -> Workflow:
    """Get a single workflow by its ID, including its full setpoints and interval combinations.

    Includes the fully populated ``parameter_group_setpoints`` and any interval
    combinations.

    !!! example
        ```python
        wf = client.workflows.get_by_id(id="WFL1")
        wf.id
        # 'WFL1'
        ```

    Parameters
    ----------
    id : WorkflowId
        The workflow ID (format ``WFL...``).

    Returns
    -------
    Workflow
        The fully populated workflow.
    """
    response = self.session.get(f"{self.base_path}/{id}")
    return Workflow(**response.json())

get_by_ids

get_by_ids(*, ids: list[WorkflowId]) -> list[Workflow]

Get multiple workflows by their IDs.

Arbitrarily long ID lists are supported. Each returned workflow includes its full setpoints. Workflows not found are omitted from the result.

Example

workflows = client.workflows.get_by_ids(ids=["WFL1", "WFL2"])
[w.name for w in workflows]
# ['Cure at 25C', 'Cure at 40C']

Parameters:

Name Type Description Default
ids list[WorkflowId]

The workflow IDs to retrieve (format WFL...).

required

Returns:

Type Description
list[Workflow]

The matching workflows. Order is not guaranteed to match the input.

Source code in src/albert/collections/workflows.py
@validate_call
def get_by_ids(self, *, ids: list[WorkflowId]) -> list[Workflow]:
    """Get multiple workflows by their IDs.

    Arbitrarily long ID lists are supported. Each returned workflow includes
    its full setpoints. Workflows not found are omitted from the result.

    !!! example
        ```python
        workflows = client.workflows.get_by_ids(ids=["WFL1", "WFL2"])
        [w.name for w in workflows]
        # ['Cure at 25C', 'Cure at 40C']
        ```

    Parameters
    ----------
    ids : list[WorkflowId]
        The workflow IDs to retrieve (format ``WFL...``).

    Returns
    -------
    list[Workflow]
        The matching workflows. Order is not guaranteed to match the input.
    """
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 300] for i in range(0, len(ids), 300)]
    return [
        Workflow(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()["Items"]
    ]

get_all

get_all(max_items: int | None = None) -> Iterator[Workflow]

Iterate over all workflows.

Workflows are usually retrieved by their IDs (via get_by_id) or created as part of building a task, so a full listing is rarely needed in production. Results are returned as a lazily paginated iterator.

Example

for wf in client.workflows.get_all(max_items=10):
    print(wf.id, wf.name)

Parameters:

Name Type Description Default
max_items int

Maximum number of workflows to return in total. If None, iterates over all workflows.

None

Yields:

Type Description
Workflow

Each workflow, fully populated.

Source code in src/albert/collections/workflows.py
def get_all(
    self,
    max_items: int | None = None,
) -> Iterator[Workflow]:
    """Iterate over all workflows.

    Workflows are usually retrieved by their IDs (via [`get_by_id`][albert.collections.workflows.WorkflowCollection.get_by_id]) or created
    as part of building a task, so a full listing is rarely needed in
    production. Results are returned as a lazily paginated iterator.

    !!! example
        ```python
        for wf in client.workflows.get_all(max_items=10):
            print(wf.id, wf.name)
        ```

    Parameters
    ----------
    max_items : int, optional
        Maximum number of workflows to return in total. If None, iterates over
        all workflows.

    Yields
    ------
    Workflow
        Each workflow, fully populated.
    """

    def deserialize(items: list[dict]) -> list[Workflow]:
        return self.get_by_ids(ids=[x["albertId"] for x in items])

    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        params={},
        session=self.session,
        deserialize=deserialize,
        max_items=max_items,
    )

search

search(
    *,
    text: str | None = None,
    status: Status | list[Status] | None = None,
    created_by: str | list[str] | None = None,
    ids: WorkflowId | list[WorkflowId] | None = None,
    data_templates: str | list[str] | None = None,
    data_template_ids: str | list[str] | None = None,
    parameter_groups: str | list[str] | None = None,
    parameter_group_ids: str | list[str] | None = None,
    parameters: str | list[str] | None = None,
    unit: str | list[str] | None = None,
    parameter_set: list[WorkflowParameterSet] | None = None,
    facet_text: str | None = None,
    facet_field: str | None = None,
    facet_list: list[str] | None = None,
    contains_field: str | list[str] | None = None,
    contains_text: str | list[str] | None = None,
    search_field: list[str] | None = None,
    source_field: list[str] | None = None,
    from_created_at: str | None = None,
    to_created_at: str | None = None,
    sort_by: str | None = None,
    order_by: OrderBy = DESCENDING,
    max_items: int | None = None,
) -> Iterator[WorkflowSearchItem]

Search for workflows matching the given filters.

This is the fast path: it returns partial (unhydrated) WorkflowSearchItem hits and is best for lookups, counts, and pulling IDs. To retrieve fully populated workflows, use get_by_id or call hydrate() on a hit. Results are returned lazily as an iterator that pages through the API on demand.

Example

for item in client.workflows.search(text="Tensile", max_items=10):
    print(item.id, item.name)

Parameters:

Name Type Description Default
text str

Free-text query.

None
status Status or list[Status]

Filter by workflow status.

None
created_by str or list[str]

Filter by creator. Accepts user display name(s) or UserId(s) (e.g. "USR4227" or "Jane Doe").

None
ids WorkflowId or list[WorkflowId]

Filter by workflow ID(s) (format WFL...).

None
data_templates str or list[str]

Filter by data template name(s).

None
data_template_ids str or list[str]

Filter by data template ID(s) (format DAT...).

None
parameter_groups str or list[str]

Filter by parameter group name(s).

None
parameter_group_ids str or list[str]

Filter by parameter group ID(s) (format PRG...).

None
parameters str or list[str]

Filter by parameter name(s). Supports inline range syntax such as name(value), name(>n), name(<n), or name(min-max).

None
unit str or list[str]

Filter by unit name(s).

None
parameter_set list[WorkflowParameterSet]

Filter by parameter constraints that bind a name, value range, and unit.

None
facet_text str

Text to match within a facet search.

None
facet_field str

Field to search within for facet filtering.

None
facet_list list[str]

Facet(s) to compute over the matching workflows. One or more of "status", "createdBy", "dataTemplate", "parameterGroup", "parameter", "unit". Retrieve computed facet groups with get_all_facets.

None
contains_field str or list[str]

Field(s) to apply a "contains" search to.

None
contains_text str or list[str]

Text value(s) for the "contains" search.

None
search_field list[str]

Restrict which fields the text query searches.

None
source_field list[str]

Restrict which fields are returned in search results.

None
from_created_at str

Only include workflows created on or after this date (ISO 8601).

None
to_created_at str

Only include workflows created on or before this date (ISO 8601).

None
sort_by str

Attribute to sort results by.

None
order_by OrderBy

The order in which to sort results. Default is DESCENDING.

DESCENDING
max_items int

Maximum number of items to return in total. If None, iterates over all matching items.

None

Returns:

Type Description
Iterator[WorkflowSearchItem]

A lazy iterator of matching partial workflows. Call hydrate() on an item to fetch its full Workflow.

Source code in src/albert/collections/workflows.py
@validate_call
def search(
    self,
    *,
    text: str | None = None,
    status: Status | list[Status] | None = None,
    created_by: str | list[str] | None = None,
    ids: WorkflowId | list[WorkflowId] | None = None,
    data_templates: str | list[str] | None = None,
    data_template_ids: str | list[str] | None = None,
    parameter_groups: str | list[str] | None = None,
    parameter_group_ids: str | list[str] | None = None,
    parameters: str | list[str] | None = None,
    unit: str | list[str] | None = None,
    parameter_set: list[WorkflowParameterSet] | None = None,
    facet_text: str | None = None,
    facet_field: str | None = None,
    facet_list: list[str] | None = None,
    contains_field: str | list[str] | None = None,
    contains_text: str | list[str] | None = None,
    search_field: list[str] | None = None,
    source_field: list[str] | None = None,
    from_created_at: str | None = None,
    to_created_at: str | None = None,
    sort_by: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    max_items: int | None = None,
) -> Iterator[WorkflowSearchItem]:
    """Search for workflows matching the given filters.

    This is the fast path: it returns partial (unhydrated)
    [`WorkflowSearchItem`][albert.resources.workflows.WorkflowSearchItem] hits and is
    best for lookups, counts, and pulling IDs. To retrieve fully populated
    workflows, use [`get_by_id`][albert.collections.workflows.WorkflowCollection.get_by_id]
    or call `hydrate()` on a hit. Results are returned lazily as an iterator
    that pages through the API on demand.

    !!! example
        ```python
        for item in client.workflows.search(text="Tensile", max_items=10):
            print(item.id, item.name)
        ```

    Parameters
    ----------
    text : str, optional
        Free-text query.
    status : Status or list[Status], optional
        Filter by workflow status.
    created_by : str or list[str], optional
        Filter by creator. Accepts user display name(s) or UserId(s) (e.g.
        ``"USR4227"`` or ``"Jane Doe"``).
    ids : WorkflowId or list[WorkflowId], optional
        Filter by workflow ID(s) (format ``WFL...``).
    data_templates : str or list[str], optional
        Filter by data template name(s).
    data_template_ids : str or list[str], optional
        Filter by data template ID(s) (format ``DAT...``).
    parameter_groups : str or list[str], optional
        Filter by parameter group name(s).
    parameter_group_ids : str or list[str], optional
        Filter by parameter group ID(s) (format ``PRG...``).
    parameters : str or list[str], optional
        Filter by parameter name(s). Supports inline range syntax such as
        ``name(value)``, ``name(>n)``, ``name(<n)``, or ``name(min-max)``.
    unit : str or list[str], optional
        Filter by unit name(s).
    parameter_set : list[WorkflowParameterSet], optional
        Filter by parameter constraints that bind a name, value range, and unit.
    facet_text : str, optional
        Text to match within a facet search.
    facet_field : str, optional
        Field to search within for facet filtering.
    facet_list : list[str], optional
        Facet(s) to compute over the matching workflows. One or more of
        ``"status"``, ``"createdBy"``, ``"dataTemplate"``, ``"parameterGroup"``,
        ``"parameter"``, ``"unit"``. Retrieve computed facet groups with
        [`get_all_facets`][albert.collections.workflows.WorkflowCollection.get_all_facets].
    contains_field : str or list[str], optional
        Field(s) to apply a "contains" search to.
    contains_text : str or list[str], optional
        Text value(s) for the "contains" search.
    search_field : list[str], optional
        Restrict which fields the text query searches.
    source_field : list[str], optional
        Restrict which fields are returned in search results.
    from_created_at : str, optional
        Only include workflows created on or after this date (ISO 8601).
    to_created_at : str, optional
        Only include workflows created on or before this date (ISO 8601).
    sort_by : str, optional
        Attribute to sort results by.
    order_by : OrderBy, optional
        The order in which to sort results. Default is ``DESCENDING``.
    max_items : int, optional
        Maximum number of items to return in total. If None, iterates over all
        matching items.

    Returns
    -------
    Iterator[WorkflowSearchItem]
        A lazy iterator of matching partial workflows. Call
        `hydrate()` on an item to fetch its full
        [`Workflow`][albert.resources.workflows.Workflow].
    """
    payload: dict[str, Any] = {
        "order": order_by,
        "text": text,
        "status": ensure_list(status),
        "createdBy": ensure_list(created_by),
        "albertId": ensure_list(ids),
        "dataTemplates": ensure_list(data_templates),
        "dataTemplateIds": ensure_list(data_template_ids),
        "parameterGroups": ensure_list(parameter_groups),
        "parameterGroupIds": ensure_list(parameter_group_ids),
        "parameters": ensure_list(parameters),
        "unit": ensure_list(unit),
        "facetText": facet_text,
        "facetField": facet_field,
        "facetList": ensure_list(facet_list),
        "containsField": ensure_list(contains_field),
        "containsText": ensure_list(contains_text),
        "searchField": ensure_list(search_field),
        "sourceField": ensure_list(source_field),
        "fromCreatedAt": from_created_at,
        "toCreatedAt": to_created_at,
        "sortBy": sort_by,
        "limit": _SEARCH_PAGE_SIZE,
    }
    if parameter_set is not None:
        payload["parameterSet"] = [
            item.model_dump(by_alias=True, mode="json", exclude_none=True)
            for item in parameter_set
        ]

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        method="POST",
        json=payload,
        max_items=max_items,
        deserialize=lambda items: [
            WorkflowSearchItem.model_validate(x)._bind_collection(self) for x in items
        ],
    )

get_all_facets

get_all_facets(
    *,
    text: str | None = None,
    status: Status | list[Status] | None = None,
    created_by: str | list[str] | None = None,
    ids: WorkflowId | list[WorkflowId] | None = None,
    data_templates: str | list[str] | None = None,
    data_template_ids: str | list[str] | None = None,
    parameter_groups: str | list[str] | None = None,
    parameter_group_ids: str | list[str] | None = None,
    parameters: str | list[str] | None = None,
    unit: str | list[str] | None = None,
    parameter_set: list[WorkflowParameterSet] | None = None,
    facet_list: list[str] | None = None,
) -> list[FacetItem]

Get the facets available for a workflow search.

Facets are the grouped, counted filter options for a query, like the refinement sidebar of a search UI (e.g. how many matching workflows fall under each parameter group or unit). Use them to build progressive filtering or to summarize a result set without fetching every item.

Example

facets = client.workflows.get_all_facets(text="Tensile")
[(f.name, len(f.value)) for f in facets]
# [('Parameter Group', 3), ('Unit', 2)]

Parameters:

Name Type Description Default
text str

Free-text query.

None
status Status or list[Status]

Filter by workflow status.

None
created_by str or list[str]

Filter by creator. Accepts user display name(s) or UserId(s) (e.g. "USR4227" or "Jane Doe").

None
ids WorkflowId or list[WorkflowId]

Filter by workflow ID(s) (format WFL...).

None
data_templates str or list[str]

Filter by data template name(s).

None
data_template_ids str or list[str]

Filter by data template ID(s) (format DAT...).

None
parameter_groups str or list[str]

Filter by parameter group name(s).

None
parameter_group_ids str or list[str]

Filter by parameter group ID(s) (format PRG...).

None
parameters str or list[str]

Filter by parameter name(s). Supports inline range syntax such as name(value), name(>n), name(<n), or name(min-max).

None
unit str or list[str]

Filter by unit name(s).

None
parameter_set list[WorkflowParameterSet]

Filter by parameter constraints that bind a name, value range, and unit.

None
facet_list list[str]

Facet(s) to compute. One or more of "status", "createdBy", "dataTemplate", "parameterGroup", "parameter", "unit". If None, the platform's default facet set is returned.

None

Returns:

Type Description
list[FacetItem]

The facet groups available for the query.

Source code in src/albert/collections/workflows.py
@validate_call
def get_all_facets(
    self,
    *,
    text: str | None = None,
    status: Status | list[Status] | None = None,
    created_by: str | list[str] | None = None,
    ids: WorkflowId | list[WorkflowId] | None = None,
    data_templates: str | list[str] | None = None,
    data_template_ids: str | list[str] | None = None,
    parameter_groups: str | list[str] | None = None,
    parameter_group_ids: str | list[str] | None = None,
    parameters: str | list[str] | None = None,
    unit: str | list[str] | None = None,
    parameter_set: list[WorkflowParameterSet] | None = None,
    facet_list: list[str] | None = None,
) -> list[FacetItem]:
    """Get the facets available for a workflow search.

    Facets are the grouped, counted filter options for a query, like the
    refinement sidebar of a search UI (e.g. how many matching workflows fall
    under each parameter group or unit). Use them to build progressive
    filtering or to summarize a result set without fetching every item.

    !!! example
        ```python
        facets = client.workflows.get_all_facets(text="Tensile")
        [(f.name, len(f.value)) for f in facets]
        # [('Parameter Group', 3), ('Unit', 2)]
        ```

    Parameters
    ----------
    text : str, optional
        Free-text query.
    status : Status or list[Status], optional
        Filter by workflow status.
    created_by : str or list[str], optional
        Filter by creator. Accepts user display name(s) or UserId(s) (e.g.
        ``"USR4227"`` or ``"Jane Doe"``).
    ids : WorkflowId or list[WorkflowId], optional
        Filter by workflow ID(s) (format ``WFL...``).
    data_templates : str or list[str], optional
        Filter by data template name(s).
    data_template_ids : str or list[str], optional
        Filter by data template ID(s) (format ``DAT...``).
    parameter_groups : str or list[str], optional
        Filter by parameter group name(s).
    parameter_group_ids : str or list[str], optional
        Filter by parameter group ID(s) (format ``PRG...``).
    parameters : str or list[str], optional
        Filter by parameter name(s). Supports inline range syntax such as
        ``name(value)``, ``name(>n)``, ``name(<n)``, or ``name(min-max)``.
    unit : str or list[str], optional
        Filter by unit name(s).
    parameter_set : list[WorkflowParameterSet], optional
        Filter by parameter constraints that bind a name, value range, and unit.
    facet_list : list[str], optional
        Facet(s) to compute. One or more of ``"status"``, ``"createdBy"``,
        ``"dataTemplate"``, ``"parameterGroup"``, ``"parameter"``, ``"unit"``.
        If None, the platform's default facet set is returned.

    Returns
    -------
    list[FacetItem]
        The facet groups available for the query.
    """
    payload: dict[str, Any] = {
        "text": text,
        "status": ensure_list(status),
        "createdBy": ensure_list(created_by),
        "albertId": ensure_list(ids),
        "dataTemplates": ensure_list(data_templates),
        "dataTemplateIds": ensure_list(data_template_ids),
        "parameterGroups": ensure_list(parameter_groups),
        "parameterGroupIds": ensure_list(parameter_group_ids),
        "parameters": ensure_list(parameters),
        "unit": ensure_list(unit),
        "facetList": ensure_list(facet_list),
        "limit": 0,  # a limit of 0 asks the platform for facets only
    }
    if parameter_set is not None:
        payload["parameterSet"] = [
            item.model_dump(by_alias=True, mode="json", exclude_none=True)
            for item in parameter_set
        ]
    payload = {k: v for k, v in payload.items() if v is not None}

    response = self.session.post(url=f"{self.base_path}/search", json=payload)
    return [FacetItem.model_validate(x) for x in response.json().get("Facets") or []]