artificial.workflows.util#

async acknowledge(message: str, title: str = 'Acknowledge Message') → Any#

Show a message that must be acknowledged before the workflow continues.

actor_reference(
id: str,
ability_types: List[Tuple[str, type]],
display_name: str = '',
) → ActorReference#
create_child_job(child, job_name: str) → str#

Creates a child job without scheduling it.

Parameters:

  • child – A non-awaited call to a workflow function. E.g. child_wf(param1, param2).

  • job_name – A string name for the child job. E.g. ‘My Child Job’.

Returns:

The ID of the child job.

Example usage#
from artificial.workflows.decorators import workflow
from artificial.workflows.util import create_child_job

@workflow('Child Workflow', 'child_wf', 'lab_9eae09e-ceed-46d1-8390-87ff7dd8586d')
@parameter('plate_barcode', {'uiTitle': 'Plate Barcode'})
async def child_wf(plate_barcode: str) -> None:
    pass

@workflow('Parent Workflow', 'parent_wf', 'lab_c9eae09e-ceed-46d1-8390-87ff7dd8586d')
async def parent_wf() -> None:
    child_job_id = create_child_job(
        child_wf(plate_barcode='F123456'),
        'Child Job'
    )
get_lab_config() → Dict[str, str]#
get_org_config() → Dict[str, str]#
get_workflow_config() → Any#
priority(priority: int)#
async prompt(
prompt_type: Type,
message: str = '',
title: str = 'Enter Data',
) → Any#

Show a dialog that prompts the user to enter data corresponding to the Type passed to the dialog. Return the entered data as the given Type.

reserve(actor_ref: str, actor_type: type, runtime_only: bool = False)#

Used in a with statement to create a spanlock over a block of actions, to limit the number of concurrent executions of that block within a lab.

Parameters:

  • actor_ref – An actor reference that defines the lock to reserve.

  • actor_type – A class associated with the spanlock name to reserve in the definition of actor_ref. The actor’s configuration must provide an ability name “exclusive_locks_<spanlock name>.

  • runtime_only – True for spanlocks that should only be reserved at runtime; False for spanlocks that should also be considered during schedule computation. The default is False for backwards compatibility; however, note that schedule-time spanlocks create significantly more overhead for the scheduler.

revocable(group_name: str)#

Used in a with statement to create a region of revocable assistants associated with the given group name. Non-assistant actions cannot appear in a revocable region.

If job A is ready to run actions in a revocable region with group name G, but job B starts running actions in a revoke region with group name G, then the revocable actions in job A will be revoked until the revoke actions of job B have completed.

Parameters:

group_name – the group name shared by related revoke and revocable regions.

revoke(group_name: str, grace_period: int)#

Used in a with statement to create a region of revoking actions associated with the given group name.

If job A is ready to run actions in a revocable region with group name G, but job B starts running actions in a revoke region with group name G, then the revocable actions in job A will be revoked until the revoke actions of job B have completed.

Parameters:

  • group_name – the group name shared by related revoke and revocable regions.

  • grace_period – the duration in seconds to allow before revoking actions. The total grace period will be the sum of the grace periods specified by all revoke regions in all jobs using the same group name.

spawn_child_job(child, job_name: str) → str#

Creates and schedules a child job.

Parameters:

  • child – A non-awaited call to a workflow function. E.g. child_wf(param1, param2).

  • job_name – A string name for the child job. E.g. ‘My Child Job’.

Returns:

The ID of the child job.

Example usage#
from artificial.workflows.decorators import workflow
from artificial.workflows.util import spawn_child_job

@workflow('Child Workflow', 'child_wf', 'lab_9eae09e-ceed-46d1-8390-87ff7dd8586d')
@parameter('plate_barcode', {'uiTitle': 'Plate Barcode'})
async def child_wf(plate_barcode: str) -> None:
    pass

@workflow('Parent Workflow', 'parent_wf', 'lab_c9eae09e-ceed-46d1-8390-87ff7dd8586d')
async def parent_wf() -> None:
    child_job_id = spawn_child_job(
        child_wf(plate_barcode='F123456'),
        'Child Job'
    )