resource_model.py

ofrak.model.resource_model

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!"
        )