Skip to content

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 
52
53
54
55
56
57
@abstractmethod
def routing_key(self) -> str:
    """
    Returns:
        A string to be used as a routing key between workers.
    """

hashed_routing_key 

hashed_routing_key() -> int

Returns:

Type Description
int

An integer to be used as a hashed routing key between workers.
Source code in dp3/common/task.py 
59
60
61
62
63
64
def hashed_routing_key(self) -> int:
    """
    Returns:
        An integer to be used as a hashed routing key between workers.
    """
    return HASH(self.routing_key())

as_message abstractmethod 

as_message() -> str

Returns:

Type Description
str

A string representation of the object.
Source code in dp3/common/task.py 
66
67
68
69
70
71
@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 
137
138
139
140
141
142
143
144
145
146
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

HASH 

HASH(key: str) -> int

Hash function used to distribute tasks to worker processes. Args: key: to be hashed Returns: last 4 bytes of MD5

Source code in dp3/common/task.py 
25
26
27
28
29
30
31
32
def HASH(key: str) -> int:
    """Hash function used to distribute tasks to worker processes.
    Args:
        key: to be hashed
    Returns:
        last 4 bytes of MD5
    """
    return int(hashlib.md5(key.encode("utf8")).hexdigest()[-4:], 16)

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 
35
36
37
38
39
40
41
42
@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)