Skip to content

resource_model.py

ofrak.model.resource_model

MutableResourceModel (ResourceModel)

add_component(self, component_id, version)

Mark that a component has run on a resource.

Source code in ofrak/model/resource_model.py
def add_component(
    self,
    component_id: bytes,
    version: int,
):
    """
    Mark that a component has run on a resource.
    """
    self.is_modified = True
    self.component_versions[component_id] = version
    self.diff.component_versions_added.add((component_id, version))

add_component_for_attributes(self, component_id, version, attribute_type)

Mark that a component has added attributes to a resource

Source code in ofrak/model/resource_model.py
def add_component_for_attributes(
    self,
    component_id: bytes,
    version: int,
    attribute_type: Type[ResourceAttributes],
):
    """
    Mark that a component has added attributes to a resource
    """
    self.components_by_attributes[attribute_type] = (component_id, version)
    self.diff.attributes_component_added.add((attribute_type, component_id, version))

ResourceAttributeDependency

__init__(self, dependent_resource_id, component_id, attributes) special

Create a dependency that says the resource dependent_resource_id has some attributes attributes (added by component_id) which depend on some information in another resource. That information is either a range of the data or some attributes of that other resource.

Parameters:

Name Type Description Default
dependent_resource_id bytes

ID of the resource which has a dependency

required
component_id bytes

Type of attributes on the resource which has a dependency

required
attributes Type[ofrak.model.resource_model.ResourceAttributes]

Component which ran on the resource to create the attributes

required
Source code in ofrak/model/resource_model.py
def __init__(
    self,
    dependent_resource_id: bytes,
    component_id: bytes,
    attributes: Type[ResourceAttributes],
):
    """
    Create a dependency that says the resource ``dependent_resource_id`` has some attributes
    ``attributes`` (added by ``component_id``) which depend on some information in another
    resource. That information is either a range of the data or some attributes of that other
    resource.

    :param dependent_resource_id: ID of the resource which has a dependency
    :param component_id: Type of attributes on the resource which has a dependency
    :param attributes: Component which ran on the resource to create the attributes
    """
    self.dependent_resource_id = dependent_resource_id
    self.component_id = component_id
    self.attributes = attributes

ResourceAttributes

DATACLASS_PARAMS

Wraps immutable attributes attached to a resource. While not enforced programmatically, only analyzers should add/replace attributes attached to a resource. Additionally, a ResourceAttributes instance also defines which component attached the attributes to a specific resource.

replace_updated(resource_attributes, updated_attributes) staticmethod

Replace the fields of resource_attributes with the updated values found in updated_attributes, returning a new object. The fields having non-None values in updated_attributes are considered to be updated and will be replaced in resource_attributes if they exist there.

Both arguments must be dataclass instances.

updated_attributes is typically a descendant of ComponentConfig.

To do

This currently assumes that values can't be updated to None, but that could happen.

Exceptions:

Type Description
TypeError

if any of resource_attributes or updated_attributes isn't a dataclass instance.

Source code in ofrak/model/resource_model.py
@staticmethod
def replace_updated(
    resource_attributes: "ResourceAttributes", updated_attributes: Any
) -> "ResourceAttributes":
    """
    Replace the fields of `resource_attributes` with the updated values found in
    `updated_attributes`, returning a new object. The fields having non-`None` values in
    `updated_attributes` are considered to be updated and will be replaced in
    `resource_attributes` if they exist there.

    Both arguments must be `dataclass` instances.

    `updated_attributes` is typically a descendant of
    [ComponentConfig][ofrak.model.component_model.ComponentConfig].

    !!! todo "To do"

           This currently assumes that values can't be updated to `None`, but that could happen.

    :raises TypeError: if any of `resource_attributes` or `updated_attributes` isn't a
        dataclass instance.
    """
    for obj in (resource_attributes, updated_attributes):
        if not (dataclasses.is_dataclass(obj) and not isinstance(obj, type)):
            raise TypeError(f"{obj.__name__} must be a dataclass instance")
    updated_fields = {
        field: val
        for field, val in dataclasses.asdict(updated_attributes).items()
        if val is not None
    }
    updated_attributes = dataclasses.replace(
        resource_attributes,
        **updated_fields,
    )
    return updated_attributes

ResourceContext (ABC)

Resource context.

ResourceIndexedAttribute (Generic)

Descriptor class for values in resource attributes which can be indexed. When a field Foo of a ResourceAttributes type A is indexed, it is possible to include an r_filter or r_sort in a query to the resource service which filters the returned resource by the value of foo each of them have.

This class should not be explicitly instantiated, instead created using the @index decorator.

For example:

@dataclass
class A(ResourceAttributes):
    x: int

    @index
    def Foo(self) -> int:
        return self.x

__init__(self, getter_func, nested_indexes=()) special

Parameters:

Name Type Description Default
getter_func Callable[[Any], ~X]

Getter function for the property

required
nested_indexes Iterable[ResourceIndexedAttribute]

Additional index types that are required to calculate the value of this index

()

Exceptions:

Type Description
TypeError

if the getter function does not have a return type annotation

TypeError

if the getter does not return an indexable type

Source code in ofrak/model/resource_model.py
def __init__(
    self,
    getter_func: Callable[[Any], X],
    nested_indexes: Iterable["ResourceIndexedAttribute"] = (),
):
    """
    :param getter_func: Getter function for the property
    :param nested_indexes: Additional index types that are required to calculate the value
    of this index

    :raises TypeError: if the getter function does not have a return type annotation
    :raises TypeError: if the getter does not return an indexable type
    """
    _validate_indexed_type(getter_func)
    self.fget: Callable[[Any], X] = getter_func
    self.attributes_owner: Optional[Type[ResourceAttributes]] = None
    self.nested_indexes = nested_indexes
    self.index_name: str = getter_func.__name__

ResourceModel

Parameters:

Name Type Description Default
id ModelIdType required
data_id ModelDataIdType required
parent_id ModelParentIdType required
tags Optional[Sequence[ResourceTag]] required
attributes Optional[ModelAttributesType] required
data_dependencies Optional[ModelAttributesType]

Stores the dependencies of other resources on specific data ranges within this resource

required
attribute_dependencies Optional[ModelAttributeDependenciesType]

Stores the dependencies of other resources on ResourceAttributes of this resource

required
component_versions Optional[ModelComponentVersionsType]

Stores the version of ComponentInterface which has been run on this resource

required
components_by_attributes Optional[ModelComponentsByAttributesType]

For each ResourceAttributes, stores the id of the ComponentInterface which was run to create it

required

SortedSet (SortedSet, MutableSet, Sequence, Generic)

Typing shim for sortedcollections.SortedSet since it still has no type annotations.

See https://github.com/grantjenks/python-sortedcontainers/pull/136.

If we want to get rid of this, we either stop using SortedSet, complete that PR and get it accepted, or wait for someone else to.

index(index_value_getter=None, *, nested_indexes=())

Create a new indexable attribute for a ResourceAttributes.

Parameters:

Name Type Description Default
index_value_getter Callable[[Any], ~X]

Method of ResourceAttributes which returns the value of the index for that instance.

None
nested_indexes Iterable[ofrak.model.resource_model.ResourceIndexedAttribute]

Additional index types that are required to calculate the value of this index.

()

Returns:

Type Description
Union[Callable[[Callable[[Any], ~X]], ofrak.model.resource_model.ResourceIndexedAttribute[~X]], ofrak.model.resource_model.ResourceIndexedAttribute[~X]]

ResourceIndexedAttribute instance

Source code in ofrak/model/resource_model.py
def index(
    index_value_getter: Callable[[Any], X] = None,
    *,
    nested_indexes: Iterable[ResourceIndexedAttribute] = (),
) -> Union[
    Callable[[Callable[[Any], X]], ResourceIndexedAttribute[X]], ResourceIndexedAttribute[X]
]:
    """
    Create a new indexable attribute for a
    [ResourceAttributes][ofrak.model.resource_model.ResourceAttributes].

    :param index_value_getter: Method of
        [ResourceAttributes][ofrak.model.resource_model.ResourceAttributes] which returns the
        value of the index for that instance.
    :param nested_indexes: Additional index types that are required to calculate the value
    of this index.

    :return: [ResourceIndexedAttribute][ofrak.model.resource_model.ResourceIndexedAttribute]
        instance
    """
    # See if we're being called as @index or @index().
    if index_value_getter is None:
        # We're called with parens.
        def wrap(_index_value_getter) -> ResourceIndexedAttribute[X]:
            return ResourceIndexedAttribute[X](_index_value_getter, nested_indexes)

        return wrap  # type: ignore

    # We're called as @index without parens.
    return ResourceIndexedAttribute[X](index_value_getter)

_validate_indexed_type(getter_func) private

Verify the getter function returns a valid indexable type - a primitive type which can be compared.

Parameters:

Name Type Description Default
getter_func Callable[[Any], ~X] required

Returns:

Type Description

Exceptions:

Type Description
TypeError

if the getter function does not have a return type annotation

TypeError

if the getter does not return an indexable type

Source code in ofrak/model/resource_model.py
def _validate_indexed_type(getter_func: Callable[[Any], X]):
    """
    Verify the getter function returns a valid indexable type - a primitive type which can be
    compared.

    :param getter_func:

    :raises TypeError: if the getter function does not have a return type annotation
    :raises TypeError: if the getter does not return an indexable type
    :return:
    """

    if not hasattr(getter_func, "__annotations__"):
        raise TypeError(
            f"Index {getter_func.__name__} must have type annotations, including return type"
        )
    annotations = getattr(getter_func, "__annotations__")
    if "return" not in annotations:
        raise TypeError(
            f"Index {getter_func.__name__} must have type annotations, including return type"
        )
    index_type = annotations["return"]
    if type(index_type) is str:
        # handles case where type annotations is "stringified" and not the actual type, e.g.
        # def foo(self) -> "int": ...
        index_type_name = index_type
    else:
        index_type_name = index_type.__name__

    if index_type_name not in _INDEXABLE_TYPES:
        raise TypeError(
            f"Type of index {getter_func.__name__} is {index_type}, which is not "
            f"one of {_INDEXABLE_TYPES.values()}; cannot index by this value!"
        )