Workflow Nodes API
WorkflowNodeCollection and WorkflowNodeLabelCollection reference — get, list, iter, count, one, first for workflow nodes and their labels.
WorkflowNodeCollection is accessed via ManagementProject.workflow_nodes. It provides the full collection contract for workflow nodes in a project. The collection also exposes a .labels attribute (WorkflowNodeLabelCollection) for querying the label taxonomy.
proj = mgmt.project("proj_grating_7f3a")
nodes = proj.workflow_nodes # WorkflowNodeCollection
labels = proj.workflow_nodes.labels # WorkflowNodeLabelCollectionThe WorkflowNode object
WorkflowNode is a frozen dataclass:
Prop
Type
The FileRef object
FileRef is a frozen dataclass, present in WorkflowNode.attachments:
Prop
Type
WorkflowNodeCollection.get(node_id, *, include=())
Fetch a single workflow node by ID. Raises if the node does not exist.
Signature
def get(self, node_id: str, *, include: Sequence[WorkflowNodeInclude] = ()) -> WorkflowNode: ...Parameters
Prop
Type
Returns — WorkflowNode.
Side effects — Makes one HTTP read request.
Raises — NotFoundError when no node with node_id exists. AuthenticationError when the API key is invalid or missing.
Example
node = proj.workflow_nodes.get(
"node_abc123",
include=("label", "properties", "parents", "children"),
)
print(node.display_name, node.label_ref.name if node.label_ref else "—")WorkflowNodeCollection.get_or_none(node_id, *, include=())
Fetch a single workflow node by ID. Returns None instead of raising when not found.
Signature
def get_or_none(self, node_id: str, *, include: Sequence[WorkflowNodeInclude] = ()) -> WorkflowNode | None: ...Parameters
Prop
Type
Returns — WorkflowNode if found, None otherwise.
Side effects — Makes one HTTP read request.
Raises — AuthenticationError when the API key is invalid or missing.
Example
node = proj.workflow_nodes.get_or_none("node_maybe")
if node:
print(node.display_name)WorkflowNodeCollection.list(...)
Return one page of workflow nodes matching the given filters.
Signature
def list(
self,
*,
label: str | None = None,
created_by: str | None = None,
has_property: dict[str, str] | None = None,
include: Sequence[WorkflowNodeInclude] = (),
limit: int = 100,
cursor: str | None = None,
order_by: WorkflowNodeOrderBy = "created_at",
order: Literal["asc", "desc"] = "desc",
) -> Page[WorkflowNode]: ...Parameters
Prop
Type
Returns — Page[WorkflowNode].
Side effects — Makes one HTTP read request.
Raises — AuthenticationError when the API key is invalid or missing.
Example
page = proj.workflow_nodes.list(
label="simulation",
has_property={"solver": "fdtd"},
include=("properties", "label"),
limit=25,
)
for node in page.items:
print(node.display_name, node.properties)WorkflowNodeCollection.iter(...)
Iterate over all workflow nodes matching the given filters across all pages.
Signature
def iter(
self,
*,
label: str | None = None,
created_by: str | None = None,
has_property: dict[str, str] | None = None,
include: Sequence[WorkflowNodeInclude] = (),
order_by: WorkflowNodeOrderBy = "created_at",
order: Literal["asc", "desc"] = "desc",
) -> Iterator[WorkflowNode]: ...Parameters
Prop
Type
Returns — Iterator[WorkflowNode]: lazy, fetches pages on demand.
Side effects — Makes one HTTP read request per page.
Raises — AuthenticationError on any page fetch.
Example
for node in proj.workflow_nodes.iter(label="measurement", include=("properties",)):
print(node.display_name, node.created_at)WorkflowNodeCollection.count(...)
Return the count of workflow nodes matching the given filters.
Signature
def count(
self,
*,
label: str | None = None,
created_by: str | None = None,
has_property: dict[str, str] | None = None,
) -> int: ...Parameters
Prop
Type
Returns — int.
Side effects — Makes one HTTP read request.
Raises — AuthenticationError when the API key is invalid or missing.
Example
total = proj.workflow_nodes.count()
fdtd_count = proj.workflow_nodes.count(has_property={"solver": "fdtd"})WorkflowNodeCollection.one(...)
Return exactly one workflow node matching the given filters.
Signature
def one(
self,
*,
label: str | None = None,
created_by: str | None = None,
has_property: dict[str, str] | None = None,
include: Sequence[WorkflowNodeInclude] = (),
) -> WorkflowNode: ...Parameters
Prop
Type
Returns — WorkflowNode.
Side effects — Makes one HTTP read request (fetches up to 2 items).
Raises — NotFoundError when no nodes match. MultipleResultsError when more than one node matches. AuthenticationError when the API key is invalid or missing.
Example
node = proj.workflow_nodes.one(has_property={"run_id": "run_2025_06_03"})WorkflowNodeCollection.first(...)
Return the first workflow node matching the given filters, or None.
Signature
def first(
self,
*,
label: str | None = None,
created_by: str | None = None,
has_property: dict[str, str] | None = None,
include: Sequence[WorkflowNodeInclude] = (),
) -> WorkflowNode | None: ...Parameters
Prop
Type
Returns — WorkflowNode if found, None otherwise. Ordered by default (created_at descending).
Side effects — Makes one HTTP read request (fetches 1 item).
Raises — AuthenticationError when the API key is invalid or missing.
Example
latest = proj.workflow_nodes.first(label="simulation")
if latest:
print(latest.display_name, latest.created_at)Node labels (workflow_nodes.labels)
WorkflowNodeCollection.labels is a WorkflowNodeLabelCollection bound to the same project. Use it to query the project's label taxonomy.
No one() or first()
WorkflowNodeLabelCollection does NOT implement one() or first(). Use get(), get_or_none(), list(), iter(), or count().
The NodeLabel object
NodeLabel is a frozen dataclass:
Prop
Type
The nested Documentation is Documentation(id, title, file, created_at, updated_at), where file is DocumentationFile(id, name, mime, size). Documentation files are private; use download_documentation to obtain a signed URL.
WorkflowNodeLabelCollection.get(label_id)
Fetch a single label by ID. Raises if not found.
Signature
def get(self, label_id: str) -> NodeLabel: ...Parameters
Prop
Type
Returns — NodeLabel.
Side effects — Makes one HTTP read request.
Raises — NotFoundError when no label with label_id exists. AuthenticationError when the API key is invalid or missing.
Example
label = proj.workflow_nodes.labels.get("lbl_fdtd_sim")
print(label.name)WorkflowNodeLabelCollection.get_or_none(label_id)
Fetch a single label by ID. Returns None if not found.
Signature
def get_or_none(self, label_id: str) -> NodeLabel | None: ...Parameters
Prop
Type
Returns — NodeLabel if found, None otherwise.
Side effects — Makes one HTTP read request.
Raises — AuthenticationError when the API key is invalid or missing.
Example
label = proj.workflow_nodes.labels.get_or_none("lbl_maybe")WorkflowNodeLabelCollection.list(...)
Return one page of labels matching the given filters.
Signature
def list(
self,
*,
name: str | None = None,
limit: int = 100,
cursor: str | None = None,
order_by: LabelOrderBy = "name",
order: Literal["asc", "desc"] = "asc",
) -> Page[NodeLabel]: ...Parameters
Prop
Type
Returns — Page[NodeLabel].
Side effects — Makes one HTTP read request.
Raises — AuthenticationError when the API key is invalid or missing.
Example
page = proj.workflow_nodes.labels.list(order_by="name")
for label in page.items:
print(label.id, label.name)WorkflowNodeLabelCollection.iter(...)
Iterate over all labels across all pages.
Signature
def iter(
self,
*,
name: str | None = None,
order_by: LabelOrderBy = "name",
order: Literal["asc", "desc"] = "asc",
) -> Iterator[NodeLabel]: ...Parameters
Prop
Type
Returns — Iterator[NodeLabel]: lazy iterator.
Side effects — Makes one HTTP read request per page.
Raises — AuthenticationError on any page fetch.
Example
for label in proj.workflow_nodes.labels.iter():
print(label.name)WorkflowNodeLabelCollection.count(*, name=None)
Return the count of labels in the project, optionally filtered by name.
Signature
def count(self, *, name: str | None = None) -> int: ...Parameters
Prop
Type
Returns — int.
Side effects — Makes one HTTP read request.
Raises — AuthenticationError when the API key is invalid or missing.
Example
n = proj.workflow_nodes.labels.count()
print(f"{n} labels defined in this project")WorkflowNodeLabelCollection.download_documentation(documentation_id)
Prepare a short-lived signed download URL for the file backing a documentation entry, when that documentation is reachable from the project (linked to a node label or to the project's workflow schema).
Signature
def download_documentation(self, documentation_id: str) -> DocumentationDownload: ...Parameters
Prop
Type
Returns — DocumentationDownload(documentation_id, file, download_url, expires_at, method).
Side effects — Makes one HTTP read request and mints a signed object-store GET URL.
Raises — NotFoundError when the documentation is not reachable from the project. AuthenticationError when the API key is invalid or missing.
Example
label = proj.workflow_nodes.labels.get("lbl_fdtd_sim")
if label.documentation:
dl = proj.workflow_nodes.labels.download_documentation(label.documentation.id)
print(dl.download_url, dl.expires_at)Members API
ManagementProject.get_all_members — paginated list of all members across all teams in a project, with optional team membership sideloading.
Audit Events API
AuditEventCollection reference — get, get_or_none, list, iter, count, one, and first for the project audit log, with time-range and level filters.