resource_model.py

ofrak.model.resource_model

Data (ResourceAttributes) dataclass

Special attributes class for accessing info about a resource's binary data. Users should never access or modify this directly! Changing the fields of this data structure will not change any about the resource's data! At best, it will do nothing, and at worst it will screw up sorting/filtering.

MutableResourceModel (ResourceModel)

__hash__(self) special

Return hash(self).

Source code in ofrak/model/resource_model.py

def __hash__(self):
    return self.id.__hash__()

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: RA, updated_attributes: Any) -> RA:
    """
    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, uses_indexes=()) special

Parameters:

Name	Type	Description	Default
getter_func	Callable[[Any], ~X]	Getter function for the property	required
uses_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],
    uses_indexes: Iterable["ResourceIndexedAttribute"] = (),
):
    """
    :param getter_func: Getter function for the property
    :param uses_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.uses_indexes = uses_indexes
    self.used_by_indexes: List["ResourceIndexedAttribute"] = []
    self.index_name: str = getter_func.__name__

    for other_index in self.uses_indexes:
        other_index.used_by_indexes.append(self)

ResourceModel

Parameters:

Name	Type	Description	Default
id	ModelIdType		required
data_id	ModelDataIdType		required
parent_id	ModelParentIdType		required
tags	Optional[Set[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

index(index_value_getter=None, *, uses_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
uses_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,
    *,
    uses_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 uses_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, uses_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!"
        )