dp3.common.task

Task

Bases: BaseModel, ABC

A generic task type class.

An abstraction for the task queue classes to depend upon.

routing_key abstractmethod

routing_key() -> str

Returns:

Type	Description
str	A string to be used as a routing key between workers.

Source code in dp3/common/task.py

@abstractmethod
def routing_key(self) -> str:
    """
    Returns:
        A string to be used as a routing key between workers.
    """

as_message abstractmethod

as_message() -> str

Returns:

Type	Description
str	A string representation of the object.

Source code in dp3/common/task.py

@abstractmethod
def as_message(self) -> str:
    """
    Returns:
        A string representation of the object.
    """

DataPointTask

DataPointTask(__pydantic_self__, **data: Any)

Bases: Task

DataPointTask

Contains single task to be pushed to TaskQueue and processed. Attributes: etype: Entity type eid: Entity id / key data_points: List of DataPoints to process tags: List of tags ttl_tokens: Dictionary of TTL tokens. delete: If True, delete entity

Set the model_spec context variable and initialize the Task.

See https://docs.pydantic.dev/latest/concepts/validators/#validation-context

Source code in dp3/common/task.py

def __init__(__pydantic_self__, **data: Any) -> None:
    """Set the `model_spec` context variable and initialize the Task.

    See https://docs.pydantic.dev/latest/concepts/validators/#validation-context
    """
    __pydantic_self__.__pydantic_validator__.validate_python(
        data,
        self_instance=__pydantic_self__,
        context=_init_context_var.get(),
    )

Snapshot

Bases: Task

Snapshot

Contains a list of entities, the meaning of which depends on the type. If type is "task", then the list contains linked entities for which a snapshot should be created. Otherwise type is "linked_entities", indicating which entities must be skipped in a parallelized creation of unlinked entities.

Attributes:

Name	Type	Description
entities	list[tuple[str, str]]	List of (entity_type, entity_id)
time	datetime	timestamp for snapshot creation
final	bool	If True, this is the last linked snapshot for the given time

task_context

task_context(model_spec: ModelSpec) -> Iterator[None]

Context manager for setting the model_spec context variable.

Source code in dp3/common/task.py

@contextmanager
def task_context(model_spec: ModelSpec) -> Iterator[None]:
    """Context manager for setting the `model_spec` context variable."""
    token = _init_context_var.set({"model_spec": model_spec})
    try:
        yield
    finally:
        _init_context_var.reset(token)