dp3.common.datatype

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

@model_validator(mode="after")
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}>"

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

    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

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