alab_management.resource_manager.resource_requester module

alab_management.resource_manager.resource_requester module#

TaskLauncher is the core module of the system, which actually executes the tasks.

exception CombinedTimeoutError[source]#

Bases: TimeoutError, TimeoutError

Combined TimeoutError.

If you catch either TimeoutError or concurrent.futures.TimeoutError, this will catch both.

class DeviceRequest(**data)[source]#

Bases: BaseModel

Pydantic model for device request.

content: str#

identifier: str#

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'content': FieldInfo(annotation=str, required=True), 'identifier': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

exception RequestCanceledError[source]#

Bases: Exception

Request Canceled Error.

class RequestMixin[source]#

Bases: object

Simple wrapper for the request collection.

get_request(request_id, **kwargs)[source]#

Get a request by request_id.

Return type:: dict[str, Any] | None

get_requests_by_status(status)[source]#: Get all requests by status.

get_requests_by_task_id(task_id)[source]#: Get all requests by task_id.

update_request_status(request_id, status, original_status=None)[source]#: Update the status of a request by request_id.

class ResourceRequestItem(**data)[source]#

Bases: BaseModel

Pydantic model for resource request item.

device: DeviceRequest#

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'device': FieldInfo(annotation=DeviceRequest, required=True), 'sample_positions': FieldInfo(annotation=list[SamplePositionRequest], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

sample_positions: list[SamplePositionRequest]#

class ResourceRequester(task_id)[source]#

Bases: RequestMixin

Class for request lab resources easily. This class will insert a request into the database, and then the task manager will read from the database and assign the resources.

It is used in LabView.

static get_concurrent_result(f, timeout=None)[source]#: Get the result of a future with a timeout. If the request is canceled, we will catch a RequestCanceledError and hang the program. The hanged program will be killed by the abort exception in the task actor, which will be handled in the task actor to clean up the lab.

release_all_resources()[source]#

Release all requests by task_id, used for error recovery.

For the requests that are not fulfilled, they will be marked as CANCELED.

For the request that have been fulfilled, they will be marked as NEED_RELEASE.

For the request that have been errored, release assigned resources.

release_resources(request_id)[source]#: Release a request by request_id.

request_resources(resource_request, timeout=None, priority=None)[source]#

Request lab resources.

Write the request into the database, and then the task manager will read from the database and assign the resources.

Return type:: dict[str, Any]

class ResourcesRequest(root=PydanticUndefined, **data)[source]#

Bases: RootModel

This class is used to validate the resource request. Each request should have a format like this.

[
    {
        "device":{
            "identifier": "name" or "type" or "nodevice",
            "content": string corresponding to identifier
        },
        "sample_positions": [
            {
                "prefix": prefix of sample position,
                "number": integer number of such positions requested.
            },
            ...
        ]
    },
    ...
].

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'root': FieldInfo(annotation=list[ResourceRequestItem], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod preprocess(values)[source]#: Preprocess the request.

root: list[ResourceRequestItem]#

alab_management.resource_manager.resource_requester module

Contents

alab_management.resource_manager.resource_requester module#