dp3.common.datatype

AnyEidT module-attribute

AnyEidT = Union[str, int, IPv4Address, IPv6Address, MACAddress]

Type alias for any of possible entity ID data types.

Note that the type is determined based on the loaded entity configuration and in most cases is only one of the options, based on what entity is being processed.

ReadOnly

Bases: BaseModel

The ReadOnly data_type is used to avoid datapoint insertion for an attribute.

DataType

Bases: RootModel

Data type container

Represents one of primitive data types:

• tag
• binary
• string
• int
• int64
• float
• ipv4
• ipv6
• mac
• time
• special
• json

or composite data type:

• link
• array
• set
• dict
• category

data_type property

data_type: Union[type, BaseModel]

Type for incoming value validation

type_info property

type_info: str

String representation of the data type, immune to whitespace changes

hashable property

hashable: bool

Whether contained data is hashable

iterable property

iterable: bool

Whether the data type is iterable

elem_type property

elem_type: DataType

if iterable, the element data type

is_link property

is_link: bool

Whether the data type is a link between entities

mirror_link property

mirror_link: bool

If is_link, whether the link is mirrored

mirror_as property

mirror_as: Union[str, None]

If mirror_link, what is the name of the mirrored attribute

link_to property

link_to: str

If is_link, the target linked entity

_determine_value_validator

_determine_value_validator()

Determines value validator (inner data_type).

Source code in dp3/common/datatype.py

def _determine_value_validator(self):
    """Determines value validator (inner `data_type`)."""
    str_type = self.root

    self._hashable = not (
        "dict" in str_type
        or "set" in str_type
        or "array" in str_type
        or "special" in str_type
        or "json" in str_type
        or "link" in str_type
    )

    if str_type in primitive_data_types:
        # Primitive type
        data_type = primitive_data_types[str_type]

    elif m := re.match(re_array, str_type):
        # Array
        element_type = m.group(1).strip()
        value_type = DataType(root=element_type)
        if not is_primitive_element_type(value_type):
            raise ValueError(f"Data type {element_type} is not supported as an array element")
        data_type = list[value_type.data_type]
        self._iterable = True
        self._elem_type = value_type

    elif m := re.match(re_set, str_type):
        # Set
        element_type = m.group(1).strip()
        value_type = DataType(root=element_type)
        if not is_primitive_element_type(value_type):
            raise ValueError(f"Data type {element_type} is not supported as a set element")
        data_type = list[value_type.data_type]  # set is not supported by MongoDB
        self._iterable = True
        self._elem_type = value_type

    elif m := re.match(re_link, str_type):
        # Link
        etype, data, mirrored = m.group("etype"), m.group("data"), m.group("mirror")
        self._link_to = etype
        self._is_link = True
        self._link_data = bool(data)
        self._mirror_link = bool(mirrored)
        self._mirror_as = mirrored if mirrored else None
        self._type_info = f"link<{etype},{data}>"

        context = get_entity_context()
        if etype not in context["entities"]:
            raise ValueError(f"Entity type '{etype}' is not defined")
        entity_spec = context["entities"][etype]

        if etype and data:
            value_type = DataType(root=data)
            data_type = create_model(
                f"Link<{data}>",
                __base__=Link,
                eid=(entity_spec.id_data_type.data_type, ...),
                data=(value_type.data_type, ...),
            )
        else:
            data_type = create_model(
                f"Link<{etype}>",
                __base__=Link,
                eid=(entity_spec.id_data_type.data_type, ...),
            )

    elif re.match(re_dict, str_type):
        # Dict
        dict_spec = {}

        key_str = str_type.split("<")[1].split(">")[0]
        key_spec = dict(item.strip().split(":") for item in key_str.split(","))

        # For each dict key
        for k, v in key_spec.items():
            if v not in primitive_data_types:
                raise ValueError(f"Data type {v} of key {k} is not supported as a dict field")

            # Optional subattribute
            k_optional = k[-1] == "?"

            # Set (type, default value) for the key
            if k_optional:
                k = k[:-1]  # Remove question mark from key
                dict_spec[k] = (Optional[primitive_data_types[v]], None)
            else:
                dict_spec[k] = (primitive_data_types[v], ...)

        # Create model for this dict
        data_type = create_model(f"{str_type}__inner", **dict_spec)
        self._type_info = (
            "dict<" + ",".join(f"{k}:{v}" for k, v in sorted(dict_spec.items())) + ">"
        )

    elif m := re.match(re_category, str_type):
        # Category
        category_type, category_values = m.group("type"), m.group("vals")

        category_type = DataType(root=category_type)
        category_values = [
            category_type._data_type(value.strip()) for value in category_values.split(",")
        ]

        data_type = Enum(
            f"Category<{category_type}>", {str(val): val for val in category_values}
        )
    else:
        raise ValueError(f"Data type '{str_type}' is not supported")

    # Set data type
    self._data_type = data_type
    # Set default type info
    if self._type_info is None:
        self._type_info = str(data_type)
    return self

determine_value_validator

determine_value_validator()

Determines value validator (inner data_type).

Source code in dp3/common/datatype.py

@model_validator(mode="after")
def determine_value_validator(self):
    """Determines value validator (inner `data_type`)."""
    return self._determine_value_validator()

get_linked_entity

get_linked_entity() -> str

Returns linked entity id. Raises ValueError if DataType is not a link.

Source code in dp3/common/datatype.py

def get_linked_entity(self) -> str:
    """Returns linked entity id. Raises ValueError if DataType is not a link."""
    try:
        return self._link_to
    except AttributeError:
        raise ValueError(f"DataType '{self}' is not a link.") from None

link_has_data

link_has_data() -> bool

Whether link has data. Raises ValueError if DataType is not a link.

Source code in dp3/common/datatype.py

def link_has_data(self) -> bool:
    """Whether link has data. Raises ValueError if DataType is not a link."""
    try:
        return self._link_data
    except AttributeError:
        raise ValueError(f"DataType '{self}' is not a link.") from None

EidDataType

Bases: DataType

Data type container for entity id

Represents one of primitive data types: - string - int - ipv4 - ipv6 - mac