Tasks and Services#

The RecordTask and RecordService objects attached to a record are documented under Base Record Models.

The models below are the supporting types, most of which are used by compute managers rather than directly by users.

class TaskClaimBody[source]#

Bases: RestModelBase

Fields#
Field	Type	Required
`compute_tags`	`list`[`str`]	Yes
`limit`	`int`	Yes
`name_data`	`ManagerName`	Yes
`programs`	`dict`[`Annotated`[`str`, `StringConstraints`(strip_whitespace=`None`, to_upper=`None`, to_lower=True, strict=`None`, min_length=`None`, max_length=`None`, pattern=`None`, ascii_only=`None`)], `list`[`str`]]	Yes

Validators#
Validator	Mode	Fields
`validate_programs`	after	`programs`
`validate_tags`	after	`compute_tags`

name_data: ManagerName#: Name information about this manager

programs: dict[Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=True, strict=None, min_length=None, max_length=None, pattern=None, ascii_only=None)], list[str]]#

Subset of programs to claim tasks for

Validated by:: validate_programs

compute_tags: list[str]#

Subset of tags to claim tasks from

Validated by:: validate_tags

limit: int#: Limit on the number of tasks to claim

classmethod validate_tags(v)[source]#

classmethod validate_programs(v)[source]#

class TaskReturnBody[source]#

Bases: RestModelBase

Fields#
Field	Type	Required	Default
`name_data`	`ManagerName`	Yes
`results_compressed`	`dict`[`int`, `Annotated`[`Any`, `BytesAnnotation`]]	Yes

name_data: ManagerName#: Name information about this manager

results_compressed: dict[int, Annotated[Any, BytesAnnotation]]#: Compressed results of the completed computations, keyed by task id

class ServiceSubtaskRecord[source]#

Bases: BaseRecord

Fields#
Field	Type	Required	Default	Alias
`comments_`	`list`[`RecordComment`] \| `None`	No	`None`	`comments`
`compute_history_`	`list`[`ComputeHistory`] \| `None`	No	`None`	`compute_history`
`created_on`	`datetime`	Yes
`creator_user`	`str` \| `None`	Yes
`description`	`str` \| `None`	No	`None`
`extras`	`dict`[`str`, `Any`]	No	`{}`
`function`	`str`	Yes
`function_kwargs`	`dict`[`str`, `Any`]	Yes
`id`	`int`	Yes
`is_service`	`bool`	Yes
`manager_name`	`str` \| `None`	Yes
`modified_on`	`datetime`	Yes
`name`	`str` \| `None`	No	`None`
`native_files_`	`dict`[`str`, `NativeFile`] \| `None`	No	`None`	`native_files`
`properties`	`dict`[`str`, `Any`] \| `None`	Yes
`record_type`	`Literal`[‘servicesubtask’]	No	`'servicesubtask'`
`required_programs`	`dict`[`str`, `Any`]	Yes
`results`	`Any`	Yes
`service_`	`RecordService` \| `None`	No	`None`	`service`
`status`	`RecordStatusEnum`	Yes
`tags`	`list`[`str`] \| `None`	No	`None`
`task_`	`RecordTask` \| `None`	No	`None`	`task`

record_type: Literal['servicesubtask']#

required_programs: dict[str, Any]#

function: str#

function_kwargs: dict[str, Any]#

results: Any#

__init__(client=None, base_url_prefix=None, **kwargs)#

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: base_url_prefix (str | None)

property children_errors: list[BaseRecord]#: Returns errored child records

property children_status: dict[RecordStatusEnum, int]#: Returns a dictionary of the status of all children of this record

property comments: list[RecordComment] | None#

property compute_history: list[ComputeHistory]#

property error: dict[str, Any] | None#

fetch_children(include=None, force_fetch=False)#

Fetches all children of this record recursively

Parameters:

include (Iterable[str] | None)
force_fetch (bool)

classmethod fetch_children_multi(records, include=None, force_fetch=False)#

Fetches all children of the given records

This tries to work efficiently, fetching larger batches of children that can span multiple records

Parameters:

records (Iterable[BaseRecord | None])
include (Iterable[str] | None)
force_fetch (bool)

get_cache_dict(**kwargs)#

Returns a dictionary of the record meant for caching

When a record is stored in the cache, it is stored without child records (which are in the cache under their own keys). This function returns a dictionary that excludes those children.

kwargs are passed directly to the pydantic dict() function.

Return type:: dict[str, Any]

classmethod get_subclass(record_type)#

Obtain a subclass of this class given its record_type

Parameters:: record_type (str)
Return type:: Type[BaseRecord]

get_waiting_reason()#

Return type:: dict[str, Any]

model_post_init(context, /)#

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self (BaseModel) – The BaseModel instance.
context (Any) – The context.

Return type:

None

property native_files: dict[str, NativeFile] | None#

property offline: bool#

propagate_client(client, base_url_prefix)#

Propagates a client and related information to this record to any fields within this record that need it

This is expected to be called from derived class propagate_client functions as well

Parameters:: base_url_prefix (str | None)

property provenance: Provenance | None#

property service: RecordService | None#

property stderr: str | None#

property stdout: str | None#

sync_to_cache(detach=False)#

Syncs this record to the cache

If detach is True, then the record will be removed from the cache

Parameters:: detach (bool)

property task: RecordTask | None#

id#

is_service#

name#

description#

tags#

properties#

extras#

status#

manager_name#

created_on#

modified_on#

creator_user#

compute_history_#

task_#

service_#

comments_#

native_files_#

class GenericTaskResult[source]#

Bases: BaseModel

Fields#
Field	Type	Required	Default
`error`	`ComputeError` \| `None`	No	`None`
`extras`	`dict`[`str`, `Any`]	No	`{}`
`id`	`int`	Yes
`provenance`	`Provenance`	Yes
`results`	`Any`	Yes
`schema_name`	`Literal`[‘qca_generic_task_result’]	No	`'qca_generic_task_result'`
`stderr`	`str` \| `None`	No	`None`
`stdout`	`str` \| `None`	No	`None`
`success`	`bool`	Yes

schema_name: Literal['qca_generic_task_result']#

id: int#

results: Any#

stdout: str | None#

stderr: str | None#

success: bool#

provenance: Provenance#

extras: dict[str, Any]#

error: ComputeError | None#