Tasks

albert.collections.tasks

BlockId module-attribute

BlockId = Annotated[str, AfterValidator(ensure_block_id)]

DataTemplateId module-attribute

DataTemplateId = Annotated[
    str, AfterValidator(ensure_datatemplate_id)
]

TaskAdapter module-attribute

TaskAdapter = TypeAdapter(TaskUnion)

TaskId module-attribute

TaskId = Annotated[str, AfterValidator(ensure_task_id)]

WorkflowId module-attribute

WorkflowId = Annotated[
    str, AfterValidator(ensure_workflow_id)
]

AlbertHTTPError

AlbertHTTPError(response: Response)

Bases: AlbertException

Base class for all erors due to HTTP responses.

Source code in src/albert/exceptions.py

def __init__(self, response: requests.Response):
    message = self._format_message(response)
    super().__init__(message)
    self.response = response

response instance-attribute

response = response

AlbertSession

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
)

Bases: Session

A session that has a base URL, which is prefixed to all request URLs.

Parameters:

Name	Type	Description	Default
base_url	str	The base URL to prefix to all requests. (e.g., "https://sandbox.albertinvent.com")	required
retries	int	The number of retries for failed requests. Defaults to 3.	None
client_credentials	ClientCredentials | None	The client credentials for programmatic authentication. Optional if token is provided.	None
token	str | None	The JWT token for authentication. Optional if client credentials are provided.	None

Methods:

Name	Description
request

Source code in src/albert/session.py

def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
):
    super().__init__()
    self.base_url = base_url
    self.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"albert-SDK V.{albert.__version__}",
        }
    )

    if token is None and client_credentials is None:
        raise ValueError("Either client credentials or token must be specified.")

    self._provided_token = token
    self._token_manager = (
        TokenManager(base_url, client_credentials) if client_credentials is not None else None
    )

    # Set up retry logic
    retries = retries if retries is not None else 3
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=0.3,
        status_forcelist=(500, 502, 503, 504, 403),
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    self.mount("http://", adapter)
    self.mount("https://", adapter)

base_url instance-attribute

base_url = base_url

request

request(
    method: str, path: str, *args, **kwargs
) -> Response

Source code in src/albert/session.py

def request(self, method: str, path: str, *args, **kwargs) -> requests.Response:
    self.headers["Authorization"] = f"Bearer {self._access_token}"
    full_url = urljoin(self.base_url, path) if not path.startswith("http") else path
    with handle_http_errors():
        response = super().request(method, full_url, *args, **kwargs)
        response.raise_for_status()
        return response

BaseCollection

BaseCollection(*, session: AlbertSession)

BaseCollection is the base class for all collection classes.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert API Session instance.	required

Source code in src/albert/collections/base.py

def __init__(self, *, session: AlbertSession):
    self.session = session

session instance-attribute

session = session

BaseTask

Bases: BaseTaggedResource

Base class for all task types. Use PropertyTask, BatchTask, or GeneralTask for specific task types.

assigned_to class-attribute instance-attribute

assigned_to: SerializeAsEntityLink[User] | None = Field(
    default=None, alias="AssignedTo"
)

category instance-attribute

category: TaskCategory

claimed_date class-attribute instance-attribute

claimed_date: str | None = Field(
    alias="claimedDate", default=None
)

closed_date class-attribute instance-attribute

closed_date: str | None = Field(
    alias="closedDate", default=None
)

completed_date class-attribute instance-attribute

completed_date: str | None = Field(
    alias="completedDate", default=None
)

due_date class-attribute instance-attribute

due_date: str | None = Field(alias='dueDate', default=None)

id class-attribute instance-attribute

id: str | None = Field(alias='albertId', default=None)

inventory_information class-attribute instance-attribute

inventory_information: list[InventoryInformation] = Field(
    alias="Inventories", default=None
)

location class-attribute instance-attribute

location: SerializeAsEntityLink[Location] | None = Field(
    default=None, alias="Location"
)

metadata class-attribute instance-attribute

metadata: dict[str, MetadataItem] = Field(
    alias="Metadata", default_factory=dict
)

name instance-attribute

name: str

notes class-attribute instance-attribute

notes: str | None = Field(default=None)

page_state class-attribute instance-attribute

page_state: PageState | None = Field(
    alias="PageState", default=None
)

parent_id class-attribute instance-attribute

parent_id: str | None = Field(
    alias="parentId", default=None
)

pass_fail class-attribute instance-attribute

pass_fail: bool | None = Field(
    alias="passOrFail", default=None
)

priority class-attribute instance-attribute

priority: TaskPriority | None = Field(default=None)

project class-attribute instance-attribute

project: (
    SerializeAsEntityLink[Project]
    | list[SerializeAsEntityLink[Project]]
    | None
) = Field(default=None, alias="Project")

result class-attribute instance-attribute

result: str | None = Field(default=None)

security_class class-attribute instance-attribute

security_class: SecurityClass | None = Field(
    alias="class", default=None
)

sources class-attribute instance-attribute

sources: list[TaskSource] | None = Field(
    default_factory=list, alias="Sources"
)

start_date class-attribute instance-attribute

start_date: str | None = Field(
    alias="startDate", default=None
)

state class-attribute instance-attribute

state: TaskState | None = Field(default=None)

HistoryEntity

Bases: str, Enum

WORKFLOW class-attribute instance-attribute

WORKFLOW = 'workflow'

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'

PropertyTask

Bases: BaseTask

Represents a batch task.

This class is used to create and manage batch tasks. It includes the base task attributes and additional attributes specific to batch tasks.

Attributes:

Name	Type	Description
name	str	The name of the batch task.
inventory_information	list[InventoryInformation]	Information about the inventory associated with the batch task.
location	SerializeAsEntityLink[Location]	The location where the batch task is performed.
parent_id	str	The ID of the parent project.
blocks	list[Block]	A list of blocks associated with the batch task.
id	(str, optional)	The ID of the batch task, by default None.
metadata	(dict[str, MetadataItem], optional)	Metadata associated with the batch task, by default an empty dictionary.
due_date	(str, optional)	The due date of the batch task. YYY-MM-DD format, by default None.
notes	(str, optional)	Notes associated with the batch task, by default None.
priority	(TaskPriority, optional)	The priority of the batch task, by default None.
assigned_to	(SerializeAsEntityLink[User], optional)	The user assigned to the batch task, by default None.
state	(TaskState, optional)	The state of the batch task, by default None.
sources	(list[TaskSource], optional)	A list of sources associated with the batch task, by default an empty list.
security_class	(SecurityClass, optional)	The security class of the batch task, by default None.
start_date	str, read only	The start date of the batch task, by default None.
claimed_date	str, read only	The claimed date of the batch task, by default None.
completed_date	str, read only	The completed date of the batch task, by default None.
closed_date	str, read only	The closed date of the batch task, by default None.

batch_task_id class-attribute instance-attribute

batch_task_id: str | None = Field(
    alias="batchTaskId", default=None
)

blocks class-attribute instance-attribute

blocks: list[Block] | None = Field(
    alias="Blocks", default=None
)

category class-attribute instance-attribute

category: Literal[PROPERTY] = PROPERTY

qc_task class-attribute instance-attribute

qc_task: bool | None = Field(alias='qcTask', default=None)

target class-attribute instance-attribute

target: str | None = Field(default=None)

TaskCategory

Bases: str, Enum

BATCH class-attribute instance-attribute

BATCH = 'Batch'

BATCH_WITH_QC class-attribute instance-attribute

BATCH_WITH_QC = 'BatchWithQC'

GENERAL class-attribute instance-attribute

GENERAL = 'General'

PROPERTY class-attribute instance-attribute

PROPERTY = 'Property'

TaskCollection

TaskCollection(*, session: AlbertSession)

Bases: BaseCollection

TaskCollection is a collection class for managing Task entities in the Albert platform.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert Session information	required

Methods:

Name	Description
add_block	Add a block to a Property task.
create	Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.
delete	Delete a task.
get_by_id	Retrieve a task by its ID.
get_history
list	Search for tasks matching the given criteria.
remove_block	Remove a block from a Property task.
update	Update a task.
update_block_workflow	Update the workflow of a specific block within a task.

Source code in src/albert/collections/tasks.py

def __init__(self, *, session: AlbertSession):
    """Initialize the TaskCollection.

    Parameters
    ----------
    session : AlbertSession
        The Albert Session information
    """
    super().__init__(session=session)
    self.base_path = f"/api/{TaskCollection._api_version}/tasks"

base_path instance-attribute

base_path = f'/api/{_api_version}/tasks'

add_block

add_block(
    *,
    task_id: TaskId,
    data_template_id: DataTemplateId,
    workflow_id: WorkflowId,
) -> None

Add a block to a Property task.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task to add the block to.	required
data_template_id	DataTemplateId	The ID of the data template to use for the block.	required
workflow_id	WorkflowId	The ID of the workflow to assign to the block.	required

Returns:

Type	Description
None	This method does not return any value.

Source code in src/albert/collections/tasks.py

@validate_call
def add_block(
    self, *, task_id: TaskId, data_template_id: DataTemplateId, workflow_id: WorkflowId
) -> None:
    """Add a block to a Property task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to add the block to.
    data_template_id : DataTemplateId
        The ID of the data template to use for the block.
    workflow_id : WorkflowId
        The ID of the workflow to assign to the block.

    Returns
    -------
    None
        This method does not return any value.

    """
    url = f"{self.base_path}/{task_id}"
    payload = [
        {
            "id": task_id,
            "data": [
                {
                    "operation": "add",
                    "attribute": "Block",
                    "newValue": [{"datId": data_template_id, "Workflow": {"id": workflow_id}}],
                }
            ],
        }
    ]
    self.session.patch(url=url, json=payload)
    return None

create

create(*, task: BaseTask) -> BaseTask

Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

Parameters:

Name	Type	Description	Default
task	BaseTask	The task object to create.	required

Returns:

Type	Description
BaseTask	The registered task object.

Source code in src/albert/collections/tasks.py

def create(self, *, task: BaseTask) -> BaseTask:
    """Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

    Parameters
    ----------
    task : BaseTask
        The task object to create.

    Returns
    -------
    BaseTask
        The registered task object.
    """
    payload = [task.model_dump(mode="json", by_alias=True, exclude_none=True)]
    url = f"{self.base_path}/multi?category={task.category.value}"
    if task.parent_id is not None:
        url = f"{url}&parentId={task.parent_id}"
    response = self.session.post(url=url, json=payload)
    task_data = response.json()[0]
    return TaskAdapter.validate_python(task_data)

delete

delete(*, id: TaskId) -> None

Delete a task.

Parameters:

Name	Type	Description	Default
id	TaskId	The ID of the task to delete.	required

Source code in src/albert/collections/tasks.py

@validate_call
def delete(self, *, id: TaskId) -> None:
    """Delete a task.

    Parameters
    ----------
    id : TaskId
        The ID of the task to delete.
    """
    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: TaskId) -> BaseTask

Retrieve a task by its ID.

Parameters:

Name	Type	Description	Default
id	TaskId	The ID of the task to retrieve.	required

Returns:

Type	Description
BaseTask	The task object with the provided ID.

Source code in src/albert/collections/tasks.py

@validate_call
def get_by_id(self, *, id: TaskId) -> BaseTask:
    """Retrieve a task by its ID.

    Parameters
    ----------
    id : TaskId
        The ID of the task to retrieve.

    Returns
    -------
    BaseTask
        The task object with the provided ID.
    """
    url = f"{self.base_path}/multi/{id}"
    response = self.session.get(url)
    return TaskAdapter.validate_python(response.json())

get_history

get_history(
    *,
    id: TaskId,
    order: OrderBy = DESCENDING,
    limit: int = 1000,
    entity: HistoryEntity | None = None,
    blockId: str | None = None,
    startKey: str | None = None,
) -> TaskHistory

Source code in src/albert/collections/tasks.py

def get_history(
    self,
    *,
    id: TaskId,
    order: OrderBy = OrderBy.DESCENDING,
    limit: int = 1000,
    entity: HistoryEntity | None = None,
    blockId: str | None = None,
    startKey: str | None = None,
) -> TaskHistory:
    params = {
        "limit": limit,
        "orderBy": OrderBy(order).value if order else None,
        "entity": entity,
        "blockId": blockId,
        "startKey": startKey,
    }
    url = f"{self.base_path}/{id}/history"
    response = self.session.get(url, params=params)
    return TaskHistory(**response.json())

list

list(
    *,
    order: OrderBy = DESCENDING,
    text: str | None = None,
    sort_by: str | None = None,
    tags: list[str] | None = None,
    task_id: list[str] | None = None,
    linked_task: list[str] | None = None,
    category: TaskCategory | None = None,
    albert_id: list[str] | None = None,
    data_template: list[str] | None = None,
    assigned_to: list[str] | None = None,
    location: list[str] | None = None,
    priority: list[str] | None = None,
    status: list[str] | None = None,
    parameter_group: list[str] | None = None,
    created_by: list[str] | None = None,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[BaseTask]

Search for tasks matching the given criteria.

Parameters:

Name	Type	Description	Default
order	OrderBy	The order in which to return results, by default OrderBy.DESCENDING	DESCENDING
text	str | None	The text to search for, by default None	None
sort_by	str | None	The attribute to sort by, by default None	None
tags	list[str] | None	The tags to search for, by default None	None
task_id	list[str] | None	The related task IDs to search for, by default None	None
linked_task	list[str] | None	The Linked Task IDs to search for, by default None	None
category	TaskCategory | None	The category of the task to search for, by default None	None
albert_id	list[str] | None	The Albert IDs to search for, by default None	None
data_template	list[str] | None	The data template IDs to search for, by default None	None
assigned_to	list[str] | None	The User IDs to search for, by default None	None
location	list[str] | None	The Locations names to search for, by default None	None
priority	list[str] | None	The Priority levels to search for, by default None	None
status	list[str] | None	The Task Statuses to search for, by default None	None
parameter_group	list[str] | None	The related Parameter Group IDs to search for, by default None	None
created_by	list[str] | None	The User IDs of the task creators to search for, by default None	None
project_id	str | None	The Project ID to search for, by default None	None

Yields:

Type	Description
Iterator[BaseTask]	An iterator of matching Task objects.

Source code in src/albert/collections/tasks.py

def list(
    self,
    *,
    order: OrderBy = OrderBy.DESCENDING,
    text: str | None = None,
    sort_by: str | None = None,
    tags: list[str] | None = None,
    task_id: list[str] | None = None,
    linked_task: list[str] | None = None,
    category: TaskCategory | None = None,
    albert_id: list[str] | None = None,
    data_template: list[str] | None = None,
    assigned_to: list[str] | None = None,
    location: list[str] | None = None,
    priority: list[str] | None = None,
    status: list[str] | None = None,
    parameter_group: list[str] | None = None,
    created_by: list[str] | None = None,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[BaseTask]:
    """Search for tasks matching the given criteria.

    Parameters
    ----------
    order : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING
    text : str | None, optional
        The text to search for, by default None
    sort_by : str | None, optional
        The attribute to sort by, by default None
    tags : list[str] | None, optional
        The tags to search for, by default None
    task_id : list[str] | None, optional
        The related task IDs to search for, by default None
    linked_task : list[str] | None, optional
        The Linked Task IDs to search for, by default None
    category : TaskCategory | None, optional
        The category of the task to search for, by default None
    albert_id : list[str] | None, optional
        The Albert IDs to search for, by default None
    data_template : list[str] | None, optional
        The data template IDs to search for, by default None
    assigned_to : list[str] | None, optional
        The User IDs to search for, by default None
    location : list[str] | None, optional
        The Locations names to search for, by default None
    priority : list[str] | None, optional
        The Priority levels to search for, by default None
    status : list[str] | None, optional
        The Task Statuses to search for, by default None
    parameter_group : list[str] | None, optional
        The related Parameter Group IDs to search for, by default None
    created_by : list[str] | None, optional
        The User IDs of the task creators to search for, by default None
    project_id : str | None, optional
        The Project ID to search for, by default None

    Yields
    ------
    Iterator[BaseTask]
        An iterator of matching Task objects.
    """

    def deserialize(items: list[dict]) -> Iterator[BaseTask]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except (
                AlbertHTTPError,
                RetryError,
            ) as e:  # some legacy poorly formed Tasks raise 500s. The allowance on Retry error to also ignore these.
                logger.warning(f"Error fetching task '{id}': {e}")

    params = {
        "limit": limit,
        "offset": offset,
        "order": OrderBy(order).value if order else None,
        "text": text,
        "sortBy": sort_by,
        "tags": tags,
        "taskId": task_id,
        "linkedTask": linked_task,
        "category": category,
        "albertId": albert_id,
        "dataTemplate": data_template,
        "assignedTo": assigned_to,
        "location": location,
        "priority": priority,
        "status": status,
        "parameterGroup": parameter_group,
        "createdBy": created_by,
        "projectId": project_id,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        deserialize=deserialize,
        params=params,
    )

remove_block

remove_block(*, task_id: TaskId, block_id: BlockId) -> None

Remove a block from a Property task.

Parameters:

Name	Type	Description	Default
task_id	str	ID of the Task to remove the block from (e.g., TASFOR1234)	required
block_id	str	ID of the Block to remove (e.g., BLK1)	required

Returns:

Type	Description
None

Source code in src/albert/collections/tasks.py

@validate_call
def remove_block(self, *, task_id: TaskId, block_id: BlockId) -> None:
    """Remove a block from a Property task.

    Parameters
    ----------
    task_id : str
        ID of the Task to remove the block from (e.g., TASFOR1234)
    block_id : str
        ID of the Block to remove (e.g., BLK1)

    Returns
    -------
    None
    """
    url = f"{self.base_path}/{task_id}"
    payload = [
        {
            "id": task_id,
            "data": [
                {
                    "operation": "delete",
                    "attribute": "Block",
                    "oldValue": [block_id],
                }
            ],
        }
    ]
    self.session.patch(url=url, json=payload)
    return None

update

update(*, task: BaseTask) -> BaseTask

Update a task.

Parameters:

Name	Type	Description	Default
task	BaseTask	The updated Task object.	required

Returns:

Type	Description
BaseTask	The updated Task object as it exists in the Albert platform.

Source code in src/albert/collections/tasks.py

def update(self, *, task: BaseTask) -> BaseTask:
    """Update a task.

    Parameters
    ----------
    task : BaseTask
        The updated Task object.

    Returns
    -------
    BaseTask
        The updated Task object as it exists in the Albert platform.
    """
    existing = self.get_by_id(id=task.id)
    patch_payload, list_metadata_updates = self._generate_adv_patch_payload(
        updated=task, existing=existing
    )
    patch_operations = patch_payload.get("data", [])

    if len(patch_operations) == 0 and len(list_metadata_updates) == 0:
        logger.info(f"Task {task.id} is already up to date")
        return task
    path = f"{self.base_path}/{task.id}"

    for datum in patch_operations:
        patch_payload = TaskPatchPayload(data=[datum], id=task.id)
        self.session.patch(
            url=path,
            json=[patch_payload.model_dump(mode="json", by_alias=True, exclude_none=True)],
        )

    # For metadata list field updates, we clear, then update
    # since duplicate attribute values are not allowed in single patch request.
    for attribute, values in list_metadata_updates.items():
        entity_links = existing.metadata.get(attribute.split(".")[1])
        old_values = [item.id if hasattr(item, "id") else item for item in entity_links]
        clear_datum = PatchDatum(
            operation=PatchOperation.DELETE, attribute=attribute, oldValue=old_values
        )
        clear_payload = TaskPatchPayload(data=[clear_datum], id=task.id)
        self.session.patch(
            url=path,
            json=[clear_payload.model_dump(mode="json", by_alias=True, exclude_none=True)],
        )
        if values:
            update_datum = PatchDatum(
                operation=PatchOperation.UPDATE,
                attribute=attribute,
                newValue=values,
                oldValue=[],
            )

            update_payload = TaskPatchPayload(data=[update_datum], id=task.id)
            self.session.patch(
                url=path,
                json=[
                    update_payload.model_dump(mode="json", by_alias=True, exclude_none=False)
                ],
            )
    return self.get_by_id(id=task.id)

update_block_workflow

update_block_workflow(
    *,
    task_id: TaskId,
    block_id: BlockId,
    workflow_id: WorkflowId,
) -> None

Update the workflow of a specific block within a task.

This method updates the workflow of a specified block within a task.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task.	required
block_id	str	The ID of the block within the task.	required
workflow_id	str	The ID of the new workflow to be assigned to the block.	required

Returns:

Type	Description
None	This method does not return any value.

Notes

• The method asserts that the retrieved task is an instance of PropertyTask.
• If the block's current workflow matches the new workflow ID, no update is performed.
• The method handles the case where the block has a default workflow named "No Parameter Group".

Source code in src/albert/collections/tasks.py

@validate_call
def update_block_workflow(
    self, *, task_id: TaskId, block_id: BlockId, workflow_id: WorkflowId
) -> None:
    """
    Update the workflow of a specific block within a task.

    This method updates the workflow of a specified block within a task.
    Parameters
    ----------
    task_id : str
        The ID of the task.
    block_id : str
        The ID of the block within the task.
    workflow_id : str
        The ID of the new workflow to be assigned to the block.

    Returns
    -------
    None
        This method does not return any value.

    Notes
    -----
    - The method asserts that the retrieved task is an instance of `PropertyTask`.
    - If the block's current workflow matches the new workflow ID, no update is performed.
    - The method handles the case where the block has a default workflow named "No Parameter Group".
    """
    url = f"{self.base_path}/{task_id}"
    task = self.get_by_id(id=task_id)
    if not isinstance(task, PropertyTask):
        logger.error(f"Task {task_id} is not an instance of PropertyTask")
        raise TypeError(f"Task {task_id} is not an instance of PropertyTask")
    for b in task.blocks:
        if b.id != block_id:
            continue
        for w in b.workflow:
            if w.name == "No Parameter Group" and len(b.workflow) > 1:
                # hardcoded default workflow
                continue
            existing_workflow_id = w.id
    if existing_workflow_id == workflow_id:
        logger.info(f"Block {block_id} already has workflow {workflow_id}")
        return None
    patch = [
        {
            "data": [
                {
                    "operation": "update",
                    "attribute": "workflow",
                    "oldValue": existing_workflow_id,
                    "newValue": workflow_id,
                    "blockId": block_id,
                }
            ],
            "id": task_id,
        }
    ]
    self.session.patch(url=url, json=patch)
    return None

TaskHistory

Bases: BaseAlbertModel

items class-attribute instance-attribute

items: list[TaskHistoryEvent] = Field(alias='Items')

TaskPatchPayload

Bases: PatchPayload

A payload for a PATCH request to update a Task.

Attributes:

Name	Type	Description
id	str	The id of the Task to be updated.

id instance-attribute

id: str