Skip to content

resource.py

ofrak.resource

Resource

Defines methods for interacting with the data and attributes of Resources, the main building block of OFRAK.

get_id(self)

Returns:

Type Description
bytes

This resource's ID

Source code in ofrak/resource.py
def get_id(self) -> bytes:
    """
    :return: This resource's ID
    """
    return self._resource.id

get_job_id(self)

Each resource belongs to a specific "job." See JobServiceInterface.

Returns:

Type Description
bytes

The ID of the job this resource belongs to

Source code in ofrak/resource.py
def get_job_id(self) -> bytes:
    """
    Each resource belongs to a specific "job." See
    [JobServiceInterface][ofrak.service.job_service_i.JobServiceInterface].

    :return: The ID of the job this resource belongs to
    """
    return self._job_id

get_data_id(self)

Each resource may have a data ID. This refers to a DataModel representing some chunk of raw binary data.

Returns:

Type Description
Optional[bytes]

The data ID associated with this resource, if it exists

Source code in ofrak/resource.py
def get_data_id(self) -> Optional[bytes]:
    """
    Each resource may have a data ID. This refers to a
    [DataModel][ofrak.model.data_model.DataModel] representing some chunk of raw binary data.

    :return: The data ID associated with this resource, if it exists
    """
    return self._resource.data_id

is_modified(self)

Check if the resource has been modified in this context and is considered "dirty".

Returns:

Type Description
bool

True if the resource is modified, False otherwise

Source code in ofrak/resource.py
def is_modified(self) -> bool:
    """
    Check if the resource has been modified in this context and is considered "dirty".
    :return: `True` if the resource is modified, `False` otherwise
    """
    return self._resource.is_modified

get_model(self)

Get the underlying model of this resource.

Returns:

Type Description
MutableResourceModel
Source code in ofrak/resource.py
def get_model(self) -> MutableResourceModel:
    """
    Get the underlying [model][ofrak.model.resource_model.ResourceModel] of this resource.
    :return:
    """
    return self._resource

get_data(self, range=None) async

A resource often represents a chunk of underlying binary data. This method returns the entire chunk by default; this can be reduced by an optional parameter.

Parameters:

Name Type Description Default
range Optional[ofrak_type.range.Range]

A range within the resource's data, relative to the resource's data itself (e.g. Range(0, 10) returns the first 10 bytes of the chunk)

None

Returns:

Type Description
bytes

The full range or a partial range of this resource's bytes

Source code in ofrak/resource.py
async def get_data(self, range: Optional[Range] = None) -> bytes:
    """
    A resource often represents a chunk of underlying binary data. This method returns the
    entire chunk by default; this can be reduced by an optional parameter.

    :param range: A range within the resource's data, relative to the resource's data itself
    (e.g. Range(0, 10) returns the first 10 bytes of the chunk)

    :return: The full range or a partial range of this resource's bytes
    """
    if self._resource.data_id is None:
        raise ValueError(
            "Resource does not have a data_id. Cannot get data from a resource with no data"
        )
    data = await self._data_service.get_data(self._resource.data_id, range)
    if range is None:
        range = Range(0, len(data))
    self._component_context.access_trackers[self._resource.id].data_accessed.add(range)
    return data

get_data_length(self) async

Returns:

Type Description
int

The length of the underlying binary data this resource represents

Source code in ofrak/resource.py
async def get_data_length(self) -> int:
    """
    :return: The length of the underlying binary data this resource represents
    """
    if self._resource.data_id is None:
        raise ValueError(
            "Resource does not have a data_id. Cannot get data length from a "
            "resource with no data."
        )
    return await self._data_service.get_data_length(self._resource.data_id)

get_data_range_within_parent(self) async

If this resource is "mapped," i.e. its underlying data is defined as a range of its parent's underlying data, this method returns the range within the parent resource's data where this resource lies. If this resource is not mapped (it is root), it returns a range starting at 0 with length 0.

Returns:

Type Description
Range

The range of the parent's data which this resource represents

Source code in ofrak/resource.py
async def get_data_range_within_parent(self) -> Range:
    """
    If this resource is "mapped," i.e. its underlying data is defined as a range of its parent's
    underlying data, this method returns the range within the parent resource's data where this
    resource lies. If this resource is not mapped (it is root), it returns a range starting at 0
    with length 0.

    :return: The range of the parent's data which this resource represents
    """
    if self._resource.data_id is None:
        raise ValueError(
            "Resource does not have a data_id. Cannot get data range from a "
            "resource with no data."
        )
    if self._resource.parent_id is None:
        return Range(0, 0)

    parent_models = list(
        await self._resource_service.get_ancestors_by_id(self._resource.id, max_count=1)
    )
    if len(parent_models) != 1:
        raise NotFoundError(f"There is no parent for resource {self._resource.id.hex()}")
    parent_model = parent_models[0]

    parent_data_id = parent_model.data_id
    if parent_data_id is None:
        return Range(0, 0)

    try:
        return await self._data_service.get_range_within_other(
            self._resource.data_id, parent_data_id
        )
    except ValueError:
        return Range(0, 0)

get_data_range_within_root(self) async

Does the same thing as get_data_range_within_parent, except the range is relative to the root.

Returns:

Type Description
Range

The range of the root node's data which this resource represents

Source code in ofrak/resource.py
async def get_data_range_within_root(self) -> Range:
    """
    Does the same thing as `get_data_range_within_parent`, except the range is relative to the
    root.

    :return: The range of the root node's data which this resource represents
    """
    if self._resource.data_id is None:
        raise ValueError(
            "Resource does not have a data_id. Cannot get data range from a "
            "resource with no data."
        )
    return await self._data_service.get_data_range_within_root(self._resource.data_id)

search_data(self, query, start=None, end=None, max_matches=None) async

Search for some data in this resource. The query may be a regex pattern (a return value of re.compile). If the query is a regex pattern, returns a tuple of pairs with both the offset of the match and the contents of the match itself. If the query is plain bytes, a list of only the match offsets are returned.

Parameters:

Name Type Description Default
query

Plain bytes to exactly match or a regex pattern to search for

required
start

Start offset in the data model to begin searching

None
end

End offset in the data model to stop searching

None

Returns:

Type Description

A tuple of offsets matching a plain bytes query, or a list of (offset, match) pairs for a regex pattern query

Source code in ofrak/resource.py
async def search_data(self, query, start=None, end=None, max_matches=None):
    """
    Search for some data in this resource. The query may be a regex pattern (a return value
    of `re.compile`). If the query is a regex pattern, returns a tuple of pairs with both the
    offset of the match and the contents of the match itself. If the query is plain bytes, a
    list of only the match offsets are returned.

    :param query: Plain bytes to exactly match or a regex pattern to search for
    :param start: Start offset in the data model to begin searching
    :param end: End offset in the data model to stop searching

    :return: A tuple of offsets matching a plain bytes query, or a list of (offset, match) pairs
    for a regex pattern query
    """
    return await self._data_service.search(self.get_data_id(), query, start, end, max_matches)

save(self) async

If this resource has been modified, update the model stored in the resource service with the local changes.

Exceptions:

Type Description
NotFoundError

If the resource service does not have a model for this resource's ID

Source code in ofrak/resource.py
async def save(self):
    """
    If this resource has been modified, update the model stored in the resource service with
    the local changes.

    :raises NotFoundError: If the resource service does not have a model for this resource's ID
    """
    await save_resources(
        (self,),
        self._resource_service,
        self._data_service,
        self._component_context,
        self._resource_context,
        self._resource_view_context,
    )

_fetch(self, resource) async private

Update the local model with the latest version from the resource service. This will fail if this resource has been modified.

Exceptions:

Type Description
InvalidStateError

If the local resource model has been modified

NotFoundError

If the resource service does not have a model for this resource's ID

Source code in ofrak/resource.py
async def _fetch(self, resource: MutableResourceModel):
    """
    Update the local model with the latest version from the resource service. This will fail
    if this resource has been modified.

    :raises InvalidStateError: If the local resource model has been modified
    :raises NotFoundError: If the resource service does not have a model for this resource's ID
    """
    if resource.is_modified and not resource.is_deleted:
        raise InvalidStateError(
            f"Cannot fetch dirty resource {resource.id.hex()} (resource "
            f"{self.get_id().hex()} attempted fetch)"
        )
    try:
        fetched_resource = await self._resource_service.get_by_id(resource.id)
    except NotFoundError:
        if (
            resource.id in self._component_context.modification_trackers
            and resource.id in self._resource_context.resource_models
        ):
            del self._resource_context.resource_models[resource.id]
        return

    resource.reset(fetched_resource)

run(self, component_type, config=None) async

Run a single component. Runs even if the component has already been run on this resource.

Parameters:

Name Type Description Default
component_type Type[ofrak.component.interface.ComponentInterface[~CC]]

The component type (may be an interface) to get and run

required
config ~CC

Optional config to pass to the component

None

Returns:

Type Description
ComponentRunResult

A ComponentRunResult containing information on resources affected by the component

Source code in ofrak/resource.py
async def run(
    self,
    component_type: Type[ComponentInterface[CC]],
    config: CC = None,
) -> ComponentRunResult:
    """
    Run a single component. Runs even if the component has already been run on this resource.

    :param component_type: The component type (may be an interface) to get and run
    :param config: Optional config to pass to the component

    :return: A ComponentRunResult containing information on resources affected by the component
    """
    job_context = self._job_context
    component_result = await self._job_service.run_component(
        JobComponentRequest(
            self._job_id,
            self._resource.id,
            component_type.get_id(),
            config,
        ),
        job_context,
    )
    for deleted_id in component_result.resources_deleted:
        if deleted_id in self._component_context.modification_trackers:
            del self._component_context.modification_trackers[deleted_id]
    await self._fetch_resources(component_result.resources_modified)
    await self._update_views(
        component_result.resources_modified, component_result.resources_deleted
    )
    return component_result

auto_run(self, components=(), blacklisted_components=(), all_unpackers=False, all_identifiers=False, all_analyzers=False, all_packers=False) async

Automatically run multiple components which may run on this resource. From an initial set of possible components to run, this set is searched for components for which the intersection of the component's targets and this resource's tags is not empty. Accepts several optional flags to expand or restrict the initial set of components.

Parameters:

Name Type Description Default
components Iterable[Type[ofrak.component.interface.ComponentInterface]]

Components to explicitly add to the initial set of components

()
blacklisted_components Iterable[Type[ofrak.component.interface.ComponentInterface]]

Components to explicitly remove to the initial set of components

()
all_unpackers bool

If true, all Unpackers are added to the initial set of components

False
all_identifiers bool

If true, all Identifiers are added to the initial set of components

False
all_analyzers bool

If true, all Analyzers are added to the initial set of components

False

Returns:

Type Description
ComponentRunResult

A ComponentRunResult containing information on resources affected by the component

Source code in ofrak/resource.py
async def auto_run(
    self,
    components: Iterable[Type[ComponentInterface]] = tuple(),
    blacklisted_components: Iterable[Type[ComponentInterface]] = tuple(),
    all_unpackers: bool = False,
    all_identifiers: bool = False,
    all_analyzers: bool = False,
    all_packers: bool = False,
) -> ComponentRunResult:
    """
    Automatically run multiple components which may run on this resource. From an initial set
    of possible components to run, this set is searched for components for which the
    intersection of the component's targets and this resource's tags is not empty. Accepts
    several optional flags to expand or restrict the initial set of components.

    :param components: Components to explicitly add to the initial set of components
    :param blacklisted_components: Components to explicitly remove to the initial set of
    components
    :param all_unpackers: If true, all Unpackers are added to the initial set of components
    :param all_identifiers: If true, all Identifiers are added to the initial set of components
    :param all_analyzers: If true, all Analyzers are added to the initial set of components

    :return: A ComponentRunResult containing information on resources affected by the component
    """
    components_result = await self._job_service.run_components(
        JobMultiComponentRequest(
            self._job_id,
            self._resource.id,
            components_allowed=tuple(c.get_id() for c in components),
            components_disallowed=tuple(c.get_id() for c in blacklisted_components),
            all_unpackers=all_unpackers,
            all_identifiers=all_identifiers,
            all_analyzers=all_analyzers,
            all_packers=all_packers,
        )
    )
    for deleted_id in components_result.resources_deleted:
        if deleted_id in self._component_context.modification_trackers:
            del self._component_context.modification_trackers[deleted_id]
    await self._fetch_resources(components_result.resources_modified)
    await self._update_views(
        components_result.resources_modified, components_result.resources_deleted
    )
    return components_result

unpack(self) async

Unpack the resource.

Returns:

Type Description
ComponentRunResult

A ComponentRunResult containing information on resources affected by the component

Source code in ofrak/resource.py
async def unpack(self) -> ComponentRunResult:
    """
    Unpack the resource.

    :return: A ComponentRunResult containing information on resources affected by the component
    """
    return await self.auto_run(all_identifiers=True, all_unpackers=True)

analyze(self, resource_attributes) async

Analyze the resource for a specific resource attribute.

Parameters:

Name Type Description Default
resource_attributes Type[~RA] required

Returns:

Type Description
~RA
Source code in ofrak/resource.py
async def analyze(self, resource_attributes: Type[RA]) -> RA:
    """
    Analyze the resource for a specific resource attribute.

    :param Type[RA] resource_attributes:

    :return:
    """
    attributes = self._check_attributes(resource_attributes)
    if attributes is None:
        await self._analyze_attributes((resource_attributes,))
        return self.get_attributes(resource_attributes)
    else:
        return attributes

identify(self) async

Run all registered identifiers on the resource, tagging it with matching resource tags.

Source code in ofrak/resource.py
async def identify(self) -> ComponentRunResult:
    """
    Run all registered identifiers on the resource, tagging it with matching resource tags.
    """
    return await self.auto_run(all_identifiers=True)

pack(self) async

Pack the resource.

Returns:

Type Description
ComponentRunResult

A ComponentRunResult containing information on resources affected by the component

Source code in ofrak/resource.py
async def pack(self) -> ComponentRunResult:
    """
    Pack the resource.

    :return: A ComponentRunResult containing information on resources affected by the component
    """
    return await self.auto_run(all_packers=True)

auto_run_recursively(self, components=(), blacklisted_components=(), blacklisted_tags=(), all_unpackers=False, all_identifiers=False, all_analyzers=False) async

Automatically run multiple components which may run on this resource or its descendents. From an initial set of possible components to run, this set is searched for components for which the intersection of the component's targets and this resource's tags is not empty. Accepts several optional flags to expand or restrict the initial set of components. After each run, compatible components from the initial set are run on any resources which have had tags added (including newly created resources). This is repeated until no new tags are added.

Parameters:

Name Type Description Default
components Iterable[Type[ofrak.component.interface.ComponentInterface]]

Components to explicitly add to the initial set of components

()
blacklisted_components Iterable[Type[ofrak.component.interface.ComponentInterface]]

Components to explicitly remove to the initial set of components

()
all_unpackers bool

If true, all Unpackers are added to the initial set of components

False
all_identifiers bool

If true, all Identifiers are added to the initial set of components

False
all_analyzers bool

If true, all Analyzers are added to the initial set of components

False

Returns:

Type Description
ComponentRunResult

A ComponentRunResult containing information on resources affected by the component

Source code in ofrak/resource.py
async def auto_run_recursively(
    self,
    components: Iterable[Type[ComponentInterface]] = tuple(),
    blacklisted_components: Iterable[Type[ComponentInterface]] = tuple(),
    blacklisted_tags: Iterable[ResourceTag] = tuple(),
    all_unpackers: bool = False,
    all_identifiers: bool = False,
    all_analyzers: bool = False,
) -> ComponentRunResult:
    """
    Automatically run multiple components which may run on this resource or its descendents.
    From an initial set of possible components to run, this set is searched for components
    for which the intersection of the component's targets and this resource's tags is not
    empty. Accepts several optional flags to expand or restrict the initial set of
    components.
    After each run, compatible components from the initial set are run on any resources which
    have had tags added (including newly created resources). This is repeated until no new
    tags are added.

    :param components: Components to explicitly add to the initial set of components
    :param blacklisted_components: Components to explicitly remove to the initial set of
    components
    :param all_unpackers: If true, all Unpackers are added to the initial set of components
    :param all_identifiers: If true, all Identifiers are added to the initial set of components
    :param all_analyzers: If true, all Analyzers are added to the initial set of components

    :return: A ComponentRunResult containing information on resources affected by the component
    """
    components_result = await self._job_service.run_components_recursively(
        JobMultiComponentRequest(
            self._job_id,
            self._resource.id,
            components_allowed=tuple(c.get_id() for c in components),
            components_disallowed=tuple(c.get_id() for c in blacklisted_components),
            all_unpackers=all_unpackers,
            all_identifiers=all_identifiers,
            all_analyzers=all_analyzers,
            tags_ignored=tuple(blacklisted_tags),
        )
    )
    await self._fetch_resources(components_result.resources_modified)
    await self._update_views(
        components_result.resources_modified, components_result.resources_deleted
    )
    return components_result

unpack_recursively(self, blacklisted_components=(), do_not_unpack=()) async

Automatically unpack this resource and recursively unpack all of its descendants. First this resource is unpacked; then, any resource which "valid" tags were added to will also be unpacked. New resources created with tags count as resources with new tags. A "valid" tag is a tag which is not explicitly ignored via the do_not_unpack argument. The unpacking will only stop when no new "valid" tags have been added in the previous iteration. This can lead to a very long unpacking process if it is totally unconstrained.

Parameters:

Name Type Description Default
blacklisted_components Iterable[Type[ofrak.component.interface.ComponentInterface]]

Components which are blocked from running during the recursive unpacking, on this resource or any descendants.

()
do_not_unpack Iterable[ofrak.model.tag_model.ResourceTag]

Do not unpack resources with this tag, and ignore these tags when checking if any new tags have been added in this iteration.

()

Returns:

Type Description
ComponentRunResult

A ComponentRunResult containing information on resources affected by the component

Source code in ofrak/resource.py
async def unpack_recursively(
    self,
    blacklisted_components: Iterable[Type[ComponentInterface]] = tuple(),
    do_not_unpack: Iterable[ResourceTag] = tuple(),
) -> ComponentRunResult:
    """
    Automatically unpack this resource and recursively unpack all of its descendants. First
    this resource is unpacked; then, any resource which "valid" tags were added to will also be
    unpacked. New resources created with tags count as resources with new tags. A "valid" tag
    is a tag which is not explicitly ignored via the ``do_not_unpack`` argument.
    The unpacking will only stop when no new "valid" tags have been added in the previous
    iteration. This can lead to a very long unpacking process if it is totally unconstrained.

    :param blacklisted_components: Components which are blocked from running during the
    recursive unpacking, on this resource or any descendants.
    :param do_not_unpack: Do not unpack resources with this tag, and ignore these tags when
    checking if any new tags have been added in this iteration.

    :return: A ComponentRunResult containing information on resources affected by the component
    """
    return await self.auto_run_recursively(
        all_identifiers=True,
        all_unpackers=True,
        blacklisted_components=blacklisted_components,
        blacklisted_tags=do_not_unpack,
    )

pack_recursively(self) async

Recursively pack the resource, starting with its descendants.

Source code in ofrak/resource.py
async def pack_recursively(self) -> ComponentRunResult:
    """
    Recursively pack the resource, starting with its descendants.
    """
    return await self._job_service.pack_recursively(self._job_id, self._resource.id)

write_to(self, destination, pack=True) async

Recursively repack resource and write data out to an arbitrary BinaryIO destination.

Parameters:

Name Type Description Default
destination BinaryIO

Destination for packed resource data

required

Returns:

Type Description
Source code in ofrak/resource.py
async def write_to(self, destination: BinaryIO, pack: bool = True):
    """
    Recursively repack resource and write data out to an arbitrary ``BinaryIO`` destination.
    :param destination: Destination for packed resource data
    :return:
    """
    if pack is True:
        await self.pack_recursively()

    destination.write(await self.get_data())

create_child(self, tags=None, attributes=None, data=None, data_range=None) async

Create a new resource as a child of this resource. This method entirely defines the child's tags and attributes. This method also defines the child's data semantics:

A child resource can either be defined in one of three ways: 1) The resource contains no data ("Dataless" resource). Not used in practice. 2) As mapping a range of its parent's data ("Mapped" resource). For example, an instruction maps a portion of its parent basic block. 3) Defining its own new, independent data ("Unmapped" resource). For example, a file extracted from a zip archive is a child of the zip archive resource, but its data does not map to some specific range of that parent archive.

By default a resource will be defined the third way (unmapped). To specify that the resource is a mapped resource, include the optional data_range parameter set to the range of the parent's data which the child maps. That is, data_range=Range(0, 10) creates a resource which maps the first 10 bytes of the parent. The optional data param defines whether to populate the new child's data. It can be used only if the data is unmapped. If the child is unmapped, the value of data still becomes that child's data, but the parent's data is unaffected. If data and data_range are both None (default), the new child is a dataless resource.

The following table sums up the possible interactions between data and data_range:

data_range param not None data_range param None
data param not None Not allowed Child unmapped, child's data set to data
data param None Child mapped, parent's data untouched Child is dataless

Parameters:

Name Type Description Default
tags Iterable[ofrak.model.tag_model.ResourceTag]

tags to add to the new child

None
attributes Iterable[ofrak.model.resource_model.ResourceAttributes]

attributes to add to the new child

None
data Optional[bytes]

The binary data for the new child. If None and data_range is None, the resource has no data. Defaults to None.

None
data_range Optional[ofrak_type.range.Range]

The range of the parent's data which the new child maps. If None ( default), the child will not map the parent's data.

None

Returns:

Type Description
Resource
Source code in ofrak/resource.py
async def create_child(
    self,
    tags: Iterable[ResourceTag] = None,
    attributes: Iterable[ResourceAttributes] = None,
    data: Optional[bytes] = None,
    data_range: Optional[Range] = None,
) -> "Resource":
    """
    Create a new resource as a child of this resource. This method entirely defines the
    child's tags and attributes. This method also defines the child's data semantics:

    A child resource can either be defined in one of three ways:
    1) The resource contains no data ("Dataless" resource). Not used in practice.
    2) As mapping a range of its parent's data ("Mapped" resource). For example, an instruction
    maps a portion of its parent basic block.
    3) Defining its own new, independent data ("Unmapped" resource). For example,
    a file extracted from a zip archive is a child of the zip archive resource, but its data
    does not map to some specific range of that parent archive.

    By default a resource will be defined the third way (unmapped). To specify that the
    resource is a mapped resource, include the optional ``data_range`` parameter set to the
    range of the parent's data which the child maps. That is, `data_range=Range(0,
    10)` creates a resource which maps the first 10 bytes of the parent.
    The optional ``data`` param defines whether to populate the new child's data. It can be used
    only if the data is unmapped. If the child is unmapped, the value of ``data``
    still becomes that child's data, but the parent's data is unaffected. If ``data`` and
    ``data_range`` are both `None` (default), the new child is a dataless resource.

    The following table sums up the possible interactions between ``data`` and ``data_range``:

    |                          | ``data_range`` param not `None`                        | ``data_range`` param `None`                  |
    |--------------------------|--------------------------------------------------------|----------------------------------------------|
    | ``data`` param not `None` | Not allowed                                            | Child unmapped, child's data set to ``data`` |
    | ``data`` param   `None`   | Child mapped, parent's data untouched                  | Child is dataless                            |

    :param tags: [tags][ofrak.model.tag_model.ResourceTag] to add to the new child
    :param attributes: [attributes][ofrak.model.resource_model.ResourceAttributes] to add to
    the new child
    :param data: The binary data for the new child. If `None` and ``data_range`` is `None`,
    the resource has no data. Defaults to `None`.
    :param data_range: The range of the parent's data which the new child maps. If `None` (
    default), the child will not map the parent's data.
    :return:
    """
    if data is not None and data_range is not None:
        raise ValueError(
            "Cannot create a child from both data and data_range. These parameters are "
            "mutually exclusive."
        )
    resource_id = self._id_service.generate_id()
    if data_range is not None:
        if self._resource.data_id is None:
            raise ValueError(
                "Cannot create a child with mapped data from a parent that doesn't have data"
            )
        data_model_id = resource_id
        await self._data_service.create_mapped(
            data_model_id,
            self._resource.data_id,
            data_range,
        )
        data_attrs = Data(data_range.start, data_range.length())
        attributes = [data_attrs, *attributes] if attributes else [data_attrs]
    elif data is not None:
        if self._resource.data_id is None:
            raise ValueError(
                "Cannot create a child with data from a parent that doesn't have data"
            )
        data_model_id = resource_id
        await self._data_service.create_root(data_model_id, data)
        data_attrs = Data(0, len(data))
        attributes = [data_attrs, *attributes] if attributes else [data_attrs]
    else:
        data_model_id = None
    resource_model = ResourceModel.create(
        resource_id,
        data_model_id,
        self._resource.id,
        tags,
        attributes,
        self._component_context.component_id,
        self._component_context.component_version,
    )
    await self._resource_service.create(resource_model)
    if self._job_context:
        resource_tracker = self._job_context.trackers[resource_model.id]
        resource_tracker.tags_added.update(resource_model.tags)
    self._component_context.mark_resource_modified(resource_id)
    self._component_context.resources_created.add(resource_model.id)
    created_resource = await self._create_resource(resource_model)
    return created_resource

create_child_from_view(self, view, data_range=None, data=None, additional_tags=(), additional_attributes=()) async

Create a new resource as a child of this resource. The new resource will have tags and attributes as defined by the view; in this way a view can act as a template to create a new resource.

The additional_tags and additional_attributes can also be used to add more tags and attributes beyond what the view contains.

This method's data and data_range parameters have the same semantics as in create_child, in short:

data_range param not None data_range param None
data param not None Child mapped, data patched into child (and parent) Child unmapped, child's data set to data
data param None Child mapped, parent's data untouched Child is dataless

See create_child documentation for details.

Parameters:

Name Type Description Default
view ~RV

A resource view to pull tags and attributes from to populate the new child

required
data_range Optional[ofrak_type.range.Range]

The range of the parent's data which the new child maps. If None ( default), the child will not map the parent's data.

None
data Optional[bytes]

The binary data for the new child. If None and data_range is None, the resource has no data. Defaults to None.

None
additional_tags Iterable[ofrak.model.tag_model.ResourceTag]

Any tags for the child in addition to those from the view

()
additional_attributes Iterable[ofrak.model.resource_model.ResourceAttributes]

Any attributes for the child in addition to those from the view

()

Returns:

Type Description
Resource
Source code in ofrak/resource.py
async def create_child_from_view(
    self,
    view: RV,
    data_range: Optional[Range] = None,
    data: Optional[bytes] = None,
    additional_tags: Iterable[ResourceTag] = (),
    additional_attributes: Iterable[ResourceAttributes] = (),
) -> "Resource":
    """
    Create a new resource as a child of this resource. The new resource will have tags and
    attributes as defined by the [view][ofrak.model.viewable_tag_model.ViewableResourceTag];
    in this way a view can act as a template to create a new resource.

    The ``additional_tags`` and ``additional_attributes`` can also be used to add more tags
    and attributes beyond what the view contains.

    This method's ``data`` and ``data_range`` parameters have the same semantics as in
    `create_child`, in short:

    |                          | ``data_range`` param not `None`                        | ``data_range`` param `None`                  |
    |--------------------------|--------------------------------------------------------|----------------------------------------------|
    | ``data`` param not `None` | Child mapped, ``data`` patched into child (and parent) | Child unmapped, child's data set to ``data`` |
    | ``data`` param   `None`   | Child mapped, parent's data untouched                  | Child is dataless                            |

    See `create_child` documentation for details.

    :param view: A [resource view][ofrak.resource_view] to pull
    [tags][ofrak.model.tag_model.ResourceTag] and
    [attributes][ofrak.model.resource_model.ResourceAttributes] from to populate the new child
    :param data_range: The range of the parent's data which the new child maps. If `None` (
    default), the child will not map the parent's data.
    :param data: The binary data for the new child. If `None` and ``data_range`` is `None`,
    the resource has no data. Defaults to `None`.
    :param additional_tags: Any [tags][ofrak.model.tag_model.ResourceTag] for the child in
    addition to those from the ``view``
    :param additional_attributes: Any
    [attributes][ofrak.model.resource_model.ResourceAttributes] for the child in addition to
    those from the ``view``
    :return:
    """
    viewable_tag: ViewableResourceTag = type(view)
    new_resource = await self.create_child(
        tags=(viewable_tag, *additional_tags),
        attributes=(*view.get_attributes_instances().values(), *additional_attributes),
        data_range=data_range,
        data=data,
    )
    return new_resource

_view_as(self, viewable_tag) private

Try to get a view without calling any analysis, to avoid as many unnecessary asyncio.gather calls as possible.

Checks cached views first for view, and if not found, then checks if the attributes needed to create the view are already present and up-to-date, and only if both of those are not found does it return an awaitable.

Source code in ofrak/resource.py
def _view_as(self, viewable_tag: Type[RV]) -> Union[RV, Awaitable[RV]]:
    """
    Try to get a view without calling any analysis, to avoid as many unnecessary
    `asyncio.gather` calls as possible.

    Checks cached views first for view, and if not found, then checks if the attributes needed
    to create the view are already present and up-to-date, and only if both of those are not
    found does it return an awaitable.
    """
    if self._resource_view_context.has_view(self.get_id(), viewable_tag):
        # First early return: View already exists in cache
        return self._resource_view_context.get_view(self.get_id(), viewable_tag)
    if not issubclass(viewable_tag, ResourceViewInterface):
        raise ValueError(
            f"Cannot get view for resource {self.get_id().hex()} of a type "
            f"{viewable_tag.__name__} because it is not a subclass of ResourceView"
        )
    if not self.has_tag(viewable_tag):
        raise ValueError(
            f"Cannot get resource {self.get_id().hex()} as view "
            f"{viewable_tag.__name__} because the resource is not tagged as a "
            f"{viewable_tag.__name__}"
        )
    composed_attrs_types = viewable_tag.composed_attributes_types
    existing_attributes = [self._check_attributes(attrs_t) for attrs_t in composed_attrs_types]
    if all(existing_attributes):
        # Second early return: All attributes needed for view are present and up-to-date
        view = viewable_tag.create(self.get_model())
        view.resource = self  # type: ignore
        self._resource_view_context.add_view(self.get_id(), view)
        return cast(RV, view)

    # Only if analysis is absolutely necessary is an awaitable created and returned
    async def finish_view_creation(
        attrs_to_analyze: Tuple[Type[ResourceAttributes], ...]
    ) -> RV:
        await self._analyze_attributes(attrs_to_analyze)
        view = viewable_tag.create(self.get_model())
        view.resource = self  # type: ignore
        self._resource_view_context.add_view(self.get_id(), view)
        return cast(RV, view)

    return finish_view_creation(
        tuple(
            attrs_t
            for attrs_t, existing in zip(composed_attrs_types, existing_attributes)
            if not existing
        )
    )

view_as(self, viewable_tag) async

Provides a specific type of view instance for this resource. The returned instance is an object which has some of the information from this same resource, however in a simpler interface. This resource instance will itself remain available through the view's .resource property.

Parameters:

Name Type Description Default
viewable_tag Type[~RV]

A ViewableResourceTag, which this resource's model must already contain

required

Returns:

Type Description
~RV

Exceptions:

Type Description
ValueError

If the model does not contain this tag, or this tag is not a ViewableResourceTag

Source code in ofrak/resource.py
async def view_as(self, viewable_tag: Type[RV]) -> RV:
    """
    Provides a specific type of view instance for this resource. The returned instance is an
    object which has some of the information from this same resource, however in a simpler
    interface. This resource instance will itself remain available through the view's
    ``.resource`` property.
    :param viewable_tag: A ViewableResourceTag, which this resource's model must already contain

    :raises ValueError: If the model does not contain this tag, or this tag is not a
    ViewableResourceTag

    :return:
    """
    view_or_create_view_task: Union[RV, Awaitable[RV]] = self._view_as(viewable_tag)
    if isawaitable(view_or_create_view_task):
        return await view_or_create_view_task
    else:
        return cast(RV, view_or_create_view_task)

add_view(self, view)

Add all the attributes composed in a view to this resource, and tag this resource with the view type. Calling this is the equivalent of making N add_attributes calls and one add_tag call (where N is the number of attributes the view is composed of).

Parameters:

Name Type Description Default
view ResourceViewInterface

An instance of a view

required
Source code in ofrak/resource.py
def add_view(self, view: ResourceViewInterface):
    """
    Add all the attributes composed in a view to this resource, and tag this resource with
    the view type. Calling this is the equivalent of making N ``add_attributes`` calls and
    one ``add_tag`` call (where N is the number of attributes the view is composed of).

    :param view: An instance of a view
    """
    for attributes in view.get_attributes_instances().values():  # type: ignore
        self.add_attributes(attributes)
    self.add_tag(type(view))

_add_tag(self, tag) private

Associate a tag with the resource. If the resource already have the provided tag, it has no effects. All parent classes of the provided tag that are tags themselves are also added.

Source code in ofrak/resource.py
def _add_tag(self, tag: ResourceTag):
    """
    Associate a tag with the resource. If the resource already have the provided tag, it has no
    effects. All parent classes of the provided tag that are tags themselves are also added.
    """
    if self._resource.has_tag(tag, False):
        return
    self._component_context.mark_resource_modified(self._resource.id)
    new_tags = self._resource.add_tag(tag)
    if self._job_context:
        resource_tracker = self._job_context.trackers[self._resource.id]
        resource_tracker.tags_added.update(new_tags)

add_tag(self, *tags)

Associate multiple tags with the resource. If the resource already have one of the provided tag, the tag is not added. All parent classes of the provided tag that are tags themselves are also added.

Source code in ofrak/resource.py
def add_tag(self, *tags: ResourceTag):
    """
    Associate multiple tags with the resource. If the resource already have one of the provided
    tag, the tag is not added. All parent classes of the provided tag that are tags themselves
    are also added.
    """
    for tag in tags:
        self._add_tag(tag)

get_tags(self, inherit=True)

Get a set of tags associated with the resource.

Source code in ofrak/resource.py
def get_tags(self, inherit: bool = True) -> Iterable[ResourceTag]:
    """
    Get a set of tags associated with the resource.
    """
    return self._resource.get_tags(inherit)

has_tag(self, tag, inherit=True)

Determine if the resource is associated with the provided tag.

Source code in ofrak/resource.py
def has_tag(self, tag: ResourceTag, inherit: bool = True) -> bool:
    """
    Determine if the resource is associated with the provided tag.
    """
    return self._resource.has_tag(tag, inherit)

get_most_specific_tags(self)

Get all tags associated with the resource from which no other tags on that resource inherit. In other words, get the resource's tags that aren't subclassed by other tags on the resource.

For example, for a resource tagged as Elf, the result would be just [Elf] instead of [Elf, Program, GenericBinary] that Resource.get_tags returns. This is because Elf inherits from Program, which inherits from GenericBinary. Even though the resource has all of those tags, the most derived class with no other derivatives is the "most specific."

Source code in ofrak/resource.py
def get_most_specific_tags(self) -> Iterable[ResourceTag]:
    """
    Get all tags associated with the resource from which no other tags on that resource
    inherit. In other words, get the resource's tags that aren't subclassed by other tags on
    the resource.

    For example, for a resource tagged as `Elf`, the result would be just `[Elf]` instead of
    `[Elf, Program, GenericBinary]` that `Resource.get_tags` returns. This is because `Elf`
    inherits from `Program`, which inherits from `GenericBinary`. Even though the resource
    has all of those tags, the most derived class with no other derivatives is the "most
    specific."
    """
    return self._resource.get_most_specific_tags()

_check_attributes(self, attributes_type) private

Try to get the current attributes.

TODO: Should we be using the version as well? The client wouldn't know the version of the component in a client-server environment. We could do that efficiently by adding a service method that list all available components (and their version)

Parameters:

Name Type Description Default
attributes_type Type[~RA]

The type of attributes to check this resource for.

required

Returns:

Type Description
Optional[~RA]

The requested attributes if they are present and up-to-date, otherwise return None.

Source code in ofrak/resource.py
def _check_attributes(self, attributes_type: Type[RA]) -> Optional[RA]:
    """
    Try to get the current attributes.

    TODO: Should we be using the version as well? The client wouldn't know the
    version of the component in a client-server environment. We could do that efficiently by
    adding a service method that list all available components (and their version)

    :param attributes_type: The type of attributes to check this resource for.

    :return: The requested attributes if they are present and up-to-date, otherwise return None.
    """
    attributes = self._resource.get_attributes(attributes_type)
    if attributes is not None:
        # Make sure that the attributes have not been invalidated
        component_id = self._resource.get_component_id_by_attributes(type(attributes))
        if component_id is not None:
            return attributes
    return None

add_attributes(self, *attributes)

Add the provided attributes to the resource. If the resource already have the provided attributes classes, they are replaced with the provided one.

Source code in ofrak/resource.py
def add_attributes(self, *attributes: ResourceAttributes):
    """
    Add the provided attributes to the resource. If the resource already have the
    provided attributes classes, they are replaced with the provided one.
    """
    for attrs in attributes:
        self._add_attributes(attrs)

has_attributes(self, attributes_type)

Check if this resource has a value for the given attributes type.

Parameters:

Name Type Description Default
attributes_type Type[ofrak.model.resource_model.ResourceAttributes] required

Returns:

Type Description
bool
Source code in ofrak/resource.py
def has_attributes(self, attributes_type: Type[ResourceAttributes]) -> bool:
    """
    Check if this resource has a value for the given attributes type.
    :param attributes_type:
    :return:
    """
    return self._resource.has_attributes(attributes_type)

get_attributes(self, attributes_type)

If this resource has attributes matching the given type, return the value of those attributes. Otherwise returns None.

Parameters:

Name Type Description Default
attributes_type Type[~RA] required

Returns:

Type Description
~RA
Source code in ofrak/resource.py
def get_attributes(self, attributes_type: Type[RA]) -> RA:
    """
    If this resource has attributes matching the given type, return the value of those
    attributes. Otherwise returns `None`.
    :param attributes_type:
    :return:
    """
    attributes = self._resource.get_attributes(attributes_type)
    if attributes is None:
        raise NotFoundError(
            f"Cannot find attributes {attributes_type} for resource {self.get_id().hex()}"
        )

    self._component_context.access_trackers[self._resource.id].attributes_accessed.add(
        attributes_type
    )
    return attributes

remove_attributes(self, attributes_type)

Remove the value of a given attributes type from this resource, if there is such a value. If the resource does not have a value for the given attributes type, do nothing.

Parameters:

Name Type Description Default
attributes_type Type[ofrak.model.resource_model.ResourceAttributes] required

Returns:

Type Description
Source code in ofrak/resource.py
def remove_attributes(self, attributes_type: Type[ResourceAttributes]):
    """
    Remove the value of a given attributes type from this resource, if there is such a value.
    If the resource does not have a value for the given attributes type, do nothing.
    :param attributes_type:
    :return:
    """
    if not self._resource.has_attributes(attributes_type):
        return
    self._set_modified()
    self._resource.remove_attributes(attributes_type)

add_component(self, component_id, version)

Mark that a component has run on this resource

Parameters:

Name Type Description Default
component_id bytes

ID of the component which ran

required
version int

Version of the component which ran

required

Returns:

Type Description
Source code in ofrak/resource.py
def add_component(
    self,
    component_id: bytes,
    version: int,
):
    """
    Mark that a component has run on this resource

    :param component_id: ID of the component which ran
    :param version: Version of the component which ran
    :return:
    """
    self._set_modified()
    self._resource.add_component(component_id, version)

add_component_for_attributes(self, component_id, version, attributes)

Mark that a component was responsible for adding some attributes to this resource.

Parameters:

Name Type Description Default
component_id bytes

ID of the component which added the attributes

required
version int

version of the component which added the attributes

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

The type of attributes which were added

required

Returns:

Type Description
Source code in ofrak/resource.py
def add_component_for_attributes(
    self,
    component_id: bytes,
    version: int,
    attributes: Type[ResourceAttributes],
):
    """
    Mark that a component was responsible for adding some attributes to this resource.
    :param component_id: ID of the component which added the attributes
    :param version: version of the component which added the attributes
    :param attributes: The type of attributes which were added
    :return:
    """
    self._set_modified()
    self._resource.add_component_for_attributes(component_id, version, attributes)

remove_component(self, component_id, attributes=None)

Remove any information that this component ran on this resource and/or added a particular type of attributes to this resource

Parameters:

Name Type Description Default
component_id bytes

ID of the component to remove information about

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

The type of attributes to remove information about

None

Returns:

Type Description
Source code in ofrak/resource.py
def remove_component(
    self,
    component_id: bytes,
    attributes: Optional[Type[ResourceAttributes]] = None,
):
    """
    Remove any information that this component ran on this resource and/or added a particular
    type of attributes to this resource
    :param component_id: ID of the component to remove information about
    :param attributes: The type of attributes to remove information about
    :return:
    """
    self._set_modified()
    self._resource.remove_component(component_id, attributes)

has_component_run(self, component_id, desired_version=None)

Check if a particular component has run on this resource

Parameters:

Name Type Description Default
component_id bytes

ID of the component to check for

required
desired_version Optional[int]

If this is not None, also check that a specific version of component ran. Defaults to None.

None

Returns:

Type Description
bool

True if a component matching component_id and desired_version ran on this resource, False otherwise. If desired_version is None, only component_id must be matched to return True.

Source code in ofrak/resource.py
def has_component_run(self, component_id: bytes, desired_version: Optional[int] = None) -> bool:
    """
    Check if a particular component has run on this resource

    :param component_id: ID of the component to check for
    :param desired_version: If this is not `None`, also check that a specific version of
    ``component`` ran. Defaults to ``None``.
    :return: `True` if a component matching ``component_id`` and ``desired_version`` ran on
    this resource, `False` otherwise. If ``desired_version`` is `None`, only ``component_id``
    must be matched to return `True`.
    """
    version = self._resource.get_component_version(component_id)
    if version is None:
        return False
    if desired_version is None:
        return True
    return version == desired_version

queue_patch(self, patch_range, data)

Replace the data within the provided range with the provided data. This operation may shrink, expand or leave untouched the resource's data. Patches are queued up to be applied, and will only be applied to the resource's data after the component this was called from exits.

Parameters:

Name Type Description Default
patch_range Range

The range of binary data in this resource to replace

required
data bytes

The bytes to replace part of this resource's data with

required

Returns:

Type Description
Source code in ofrak/resource.py
def queue_patch(
    self,
    patch_range: Range,
    data: bytes,
):
    """
    Replace the data within the provided range with the provided data. This operation may
    shrink, expand or leave untouched the resource's data. Patches are queued up to be
    applied, and will only be applied to the resource's data after the component this was
    called from exits.

    :param patch_range: The range of binary data in this resource to replace
    :param data: The bytes to replace part of this resource's data with
    :return:
    """
    if not self._component_context:
        raise InvalidStateError(
            f"Cannot patch resource {self._resource.id.hex()} without a context"
        )
    if self._resource.data_id is None:
        raise ValueError("Cannot patch a resource with no data")
    self._component_context.modification_trackers[self._resource.id].data_patches.append(
        DataPatch(
            patch_range,
            self._resource.data_id,
            data,
        )
    )
    self._resource.is_modified = True

get_parent_as_view(self, v_type) async

Get the parent of this resource. The parent will be returned as an instance of the given viewable tag.

Parameters:

Name Type Description Default
v_type Type[~RV]

The type of view to get the parent as

required
Source code in ofrak/resource.py
async def get_parent_as_view(self, v_type: Type[RV]) -> RV:
    """
    Get the parent of this resource. The parent will be returned as an instance of the given
    [viewable tag][ofrak.model.viewable_tag_model.ViewableResourceTag].

    :param v_type: The type of [view][ofrak.resource] to get the parent as
    """
    parent_r = await self.get_parent()
    return await parent_r.view_as(v_type)

get_parent(self) async

Get the parent of this resource.

Source code in ofrak/resource.py
async def get_parent(self) -> "Resource":
    """
    Get the parent of this resource.
    """
    models = list(
        await self._resource_service.get_ancestors_by_id(self._resource.id, max_count=1)
    )
    if len(models) != 1:
        raise NotFoundError(f"There is no parent for resource {self._resource.id.hex()}")
    return await self._create_resource(models[0])

get_ancestors(self, r_filter=None) async

Get all the ancestors of this resource. May optionally filter the ancestors so only those matching certain parameters are returned.

Parameters:

Name Type Description Default
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None

Returns:

Type Description
Iterable[Resource]

Exceptions:

Type Description
NotFoundError

If a filter was provided and no resources match the provided filter

Source code in ofrak/resource.py
async def get_ancestors(
    self,
    r_filter: ResourceFilter = None,
) -> Iterable["Resource"]:
    """
    Get all the ancestors of this resource. May optionally filter the ancestors so only those
    matching certain parameters are returned.

    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If a filter was provided and no resources match the provided filter
    """
    models = await self._resource_service.get_ancestors_by_id(
        self._resource.id, r_filter=r_filter
    )
    return await self._create_resources(models)

get_only_ancestor_as_view(self, v_type, r_filter) async

Get the only ancestor of this resource which matches the given filter. The ancestor will be returned as an instance of the given viewable tag.

Parameters:

Name Type Description Default
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

required

Returns:

Type Description
~RV

Exceptions:

Type Description
NotFoundError

If more or fewer than one ancestor matches r_filter

Source code in ofrak/resource.py
async def get_only_ancestor_as_view(
    self,
    v_type: Type[RV],
    r_filter: ResourceFilter,
) -> RV:
    """
    Get the only ancestor of this resource which matches the given filter. The ancestor will be
    returned as an instance of the given
    [viewable tag][ofrak.model.viewable_tag_model.ViewableResourceTag].

    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If more or fewer than one ancestor matches ``r_filter``
    """
    ancestor_r = await self.get_only_ancestor(r_filter)
    return await ancestor_r.view_as(v_type)

get_only_ancestor(self, r_filter) async

Get the only ancestor of this resource which matches the given filter.

Parameters:

Name Type Description Default
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

required

Returns:

Type Description
Resource
Source code in ofrak/resource.py
async def get_only_ancestor(self, r_filter: ResourceFilter) -> "Resource":
    """
    Get the only ancestor of this resource which matches the given filter.

    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:
    """
    ancestors = list(
        await self._resource_service.get_ancestors_by_id(self._resource.id, 1, r_filter)
    )
    if len(ancestors) == 0:
        raise NotFoundError(
            f"There is no ancestor for resource {self._resource.id.hex()} matching the "
            f"provided filter"
        )
    return await self._create_resource(ancestors[0])

get_descendants_as_view(self, v_type, max_depth=-1, r_filter=None, r_sort=None) async

Get all the descendants of this resource. May optionally filter the descendants so only those matching certain parameters are returned. May optionally sort the descendants by an indexable attribute value key. The descendants will be returned as an instance of the given viewable tag.

Parameters:

Name Type Description Default
v_type Type[~RV]

The type of view to get the descendants as

required
max_depth int

Maximum depth from this resource to search for descendants; if -1, no maximum depth

-1
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None
r_sort ResourceSort

Specifies which indexable attribute to use as the key to sort and the direction to sort

None

Returns:

Type Description
Iterable[~RV]

Exceptions:

Type Description
NotFoundError

If a filter was provided and no resources match the provided filter

Source code in ofrak/resource.py
async def get_descendants_as_view(
    self,
    v_type: Type[RV],
    max_depth: int = -1,
    r_filter: ResourceFilter = None,
    r_sort: ResourceSort = None,
) -> Iterable[RV]:
    """
    Get all the descendants of this resource. May optionally filter the descendants so only
    those matching certain parameters are returned. May optionally sort the descendants by
    an indexable attribute value key. The descendants will be returned as an
    instance of the given [viewable tag][ofrak.model.viewable_tag_model.ViewableResourceTag].

    :param v_type: The type of [view][ofrak.resource] to get the descendants as
    :param max_depth: Maximum depth from this resource to search for descendants; if -1,
    no maximum depth
    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :param r_sort: Specifies which indexable attribute to use as the key to sort and the
    direction to sort
    :return:

    :raises NotFoundError: If a filter was provided and no resources match the provided filter
    """
    descendants = await self.get_descendants(max_depth, r_filter, r_sort)
    views_or_tasks = [r._view_as(v_type) for r in descendants]
    # analysis tasks to generate views of resources which don't have attrs for the view already
    view_tasks: List[Awaitable[RV]] = []
    # each resources' already-existing views OR the index in `view_tasks` of the analysis task
    views_or_task_indexes: List[Union[int, RV]] = []
    for view_or_create_view_task in views_or_tasks:
        if isawaitable(view_or_create_view_task):
            views_or_task_indexes.append(len(view_tasks))
            view_tasks.append(view_or_create_view_task)
        else:
            views_or_task_indexes.append(cast(RV, view_or_create_view_task))

    if view_tasks:
        completed_views: Sequence[RV] = await asyncio.gather(*view_tasks)
        return [
            completed_views[v_or_i] if type(v_or_i) is int else cast(RV, v_or_i)
            for v_or_i in views_or_task_indexes
        ]
    else:
        # There are no tasks, so all needed views are already present
        return cast(List[RV], views_or_task_indexes)

get_descendants(self, max_depth=-1, r_filter=None, r_sort=None) async

Get all the descendants of this resource. May optionally filter the descendants so only those matching certain parameters are returned. May optionally sort the descendants by an indexable attribute value key.

Parameters:

Name Type Description Default
max_depth int

Maximum depth from this resource to search for descendants; if -1, no maximum depth

-1
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None
r_sort ResourceSort

Specifies which indexable attribute to use as the key to sort and the direction to sort

None

Returns:

Type Description
Iterable[Resource]

Exceptions:

Type Description
NotFoundError

If a filter was provided and no resources match the provided filter

Source code in ofrak/resource.py
async def get_descendants(
    self,
    max_depth: int = -1,
    r_filter: ResourceFilter = None,
    r_sort: ResourceSort = None,
) -> Iterable["Resource"]:
    """
    Get all the descendants of this resource. May optionally filter the descendants so only
    those matching certain parameters are returned. May optionally sort the descendants by
    an indexable attribute value key.

    :param max_depth: Maximum depth from this resource to search for descendants; if -1,
    no maximum depth
    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :param r_sort: Specifies which indexable attribute to use as the key to sort and the
    direction to sort
    :return:

    :raises NotFoundError: If a filter was provided and no resources match the provided filter
    """
    models = await self._resource_service.get_descendants_by_id(
        self._resource.id, max_depth=max_depth, r_filter=r_filter, r_sort=r_sort
    )
    return await self._create_resources(models)

get_only_descendant_as_view(self, v_type, max_depth=-1, r_filter=None) async

If a filter is provided, get the only descendant of this resource which matches the given filter. If a filter is not provided, gets the only descendant of this resource. The descendant will be returned as an instance of the given viewable tag.

Parameters:

Name Type Description Default
v_type Type[~RV]

The type of view to get the descendant as

required
max_depth int

Maximum depth from this resource to search for descendants; if -1, no maximum depth

-1
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None

Returns:

Type Description
~RV

Exceptions:

Type Description
NotFoundError

If a filter is provided and more or fewer than one descendant matches r_filter

NotFoundError

If a filter is not provided and this resource has multiple descendant

Source code in ofrak/resource.py
async def get_only_descendant_as_view(
    self,
    v_type: Type[RV],
    max_depth: int = -1,
    r_filter: ResourceFilter = None,
) -> RV:
    """
    If a filter is provided, get the only descendant of this resource which matches the given
    filter. If a filter is not provided, gets the only descendant of this resource. The
    descendant will be returned as an instance of the given
    [viewable tag][ofrak.model.viewable_tag_model.ViewableResourceTag].

    :param v_type: The type of [view][ofrak.resource] to get the descendant as
    :param max_depth: Maximum depth from this resource to search for descendants; if -1,
    no maximum depth
    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If a filter is provided and more or fewer than one descendant matches
    ``r_filter``
    :raises NotFoundError: If a filter is not provided and this resource has multiple descendant
    """
    descendant_r = await self.get_only_descendant(max_depth, r_filter)
    return await descendant_r.view_as(v_type)

get_only_descendant(self, max_depth=-1, r_filter=None) async

If a filter is provided, get the only descendant of this resource which matches the given filter. If a filter is not provided, gets the only descendant of this resource.

Parameters:

Name Type Description Default
max_depth int

Maximum depth from this resource to search for descendants; if -1, no maximum depth

-1
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None

Returns:

Type Description
Resource

Exceptions:

Type Description
NotFoundError

If a filter is provided and more or fewer than one descendant matches r_filter

NotFoundError

If a filter is not provided and this resource has multiple descendant

Source code in ofrak/resource.py
async def get_only_descendant(
    self,
    max_depth: int = -1,
    r_filter: ResourceFilter = None,
) -> "Resource":
    """
    If a filter is provided, get the only descendant of this resource which matches the given
    filter. If a filter is not provided, gets the only descendant of this resource.

    :param max_depth: Maximum depth from this resource to search for descendants; if -1,
    no maximum depth
    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If a filter is provided and more or fewer than one descendant matches
    ``r_filter``
    :raises NotFoundError: If a filter is not provided and this resource has multiple descendant
    """
    models = list(
        await self._resource_service.get_descendants_by_id(
            self._resource.id,
            max_depth=max_depth,
            max_count=2,
            r_filter=r_filter,
        )
    )
    if len(models) == 0:
        raise NotFoundError(
            f"There is no descendant for resource {self._resource.id.hex()} matching "
            f"the provided filter {r_filter}"
        )
    if len(models) > 1:
        # TODO: Not the right kind of error
        raise NotFoundError(
            f"There are multiple descendants for resource {self._resource.id.hex()} "
            f"matching the provided filter"
        )
    return await self._create_resource(models[0])

get_only_sibling_as_view(self, v_type, r_filter=None) async

If a filter is provided, get the only sibling of this resource which matches the given filter. If a filter is not provided, gets the only sibling of this resource. The sibling will be returned as an instance of the given viewable tag.

Parameters:

Name Type Description Default
v_type Type[~RV]

The type of view to get the sibling as

required
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None

Returns:

Type Description
~RV

Exceptions:

Type Description
NotFoundError

If a filter is provided and more or fewer than one sibling matches r_filter

NotFoundError

If a filter is not provided and this resource has multiple siblings

Source code in ofrak/resource.py
async def get_only_sibling_as_view(
    self,
    v_type: Type[RV],
    r_filter: ResourceFilter = None,
) -> RV:
    """
    If a filter is provided, get the only sibling of this resource which matches the given
    filter. If a filter is not provided, gets the only sibling of this resource. The sibling
    will be returned as an instance of the given
    [viewable tag][ofrak.model.viewable_tag_model.ViewableResourceTag].
    :param v_type: The type of [view][ofrak.resource] to get the sibling as
    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If a filter is provided and more or fewer than one sibling matches
    ``r_filter``
    :raises NotFoundError: If a filter is not provided and this resource has multiple siblings
    """
    sibling_r = await self.get_only_sibling(r_filter)
    return await sibling_r.view_as(v_type)

get_only_sibling(self, r_filter=None) async

If a filter is provided, get the only sibling of this resource which matches the given filter. If a filter is not provided, gets the only sibling of this resource.

Parameters:

Name Type Description Default
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None

Returns:

Type Description
Resource

Exceptions:

Type Description
NotFoundError

If a filter is provided and more or fewer than one sibling matches r_filter

NotFoundError

If a filter is not provided and this resource has multiple siblings

Source code in ofrak/resource.py
async def get_only_sibling(self, r_filter: ResourceFilter = None) -> "Resource":
    """
    If a filter is provided, get the only sibling of this resource which matches the given
    filter. If a filter is not provided, gets the only sibling of this resource.

    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If a filter is provided and more or fewer than one sibling matches
    ``r_filter``
    :raises NotFoundError: If a filter is not provided and this resource has multiple siblings
    """
    models = list(
        await self._resource_service.get_siblings_by_id(
            self._resource.id,
            max_count=2,
            r_filter=r_filter,
        )
    )
    if len(models) == 0:
        raise NotFoundError(
            f"There is no sibling for resource {self._resource.id.hex()} matching "
            f"the provided filter"
        )
    if len(models) > 1:
        raise NotFoundError(
            f"There are multiple siblings for resource {self._resource.id.hex()} "
            f"matching the provided filter"
        )
    return await self._create_resource(models[0])

get_children(self, r_filter=None, r_sort=None) async

Get all the children of this resource. May optionally sort the children by an indexable attribute value key. May optionally filter the children so only those matching certain parameters are returned.

Parameters:

Name Type Description Default
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None
r_sort ResourceSort

Specifies which indexable attribute to use as the key to sort and the direction to sort

None

Returns:

Type Description
Iterable[Resource]

Exceptions:

Type Description
NotFoundError

If a filter was provided and no resources match the provided filter

Source code in ofrak/resource.py
async def get_children(
    self,
    r_filter: ResourceFilter = None,
    r_sort: ResourceSort = None,
) -> Iterable["Resource"]:
    """
    Get all the children of this resource. May optionally sort the children by an
    indexable attribute value key. May optionally filter the children so only those
    matching certain parameters are returned.

    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :param r_sort: Specifies which indexable attribute to use as the key to sort and the
    direction to sort
    :return:

    :raises NotFoundError: If a filter was provided and no resources match the provided filter
    """
    return await self.get_descendants(1, r_filter, r_sort)

get_children_as_view(self, v_type, r_filter=None, r_sort=None) async

Get all the children of this resource. May optionally filter the children so only those matching certain parameters are returned. May optionally sort the children by an indexable attribute value key. The children will be returned as an instance of the given viewable tag.

Parameters:

Name Type Description Default
v_type Type[~RV]

The type of view to get the children as

required
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None
r_sort ResourceSort

Specifies which indexable attribute to use as the key to sort and the direction to sort

None

Returns:

Type Description
Iterable[~RV]

Exceptions:

Type Description
NotFoundError

If a filter was provided and no resources match the provided filter

Source code in ofrak/resource.py
async def get_children_as_view(
    self,
    v_type: Type[RV],
    r_filter: ResourceFilter = None,
    r_sort: ResourceSort = None,
) -> Iterable[RV]:
    """
    Get all the children of this resource. May optionally filter the children so only those
    matching certain parameters are returned. May optionally sort the children by an
    indexable attribute value key. The children will be returned as an instance of
    the given [viewable tag][ofrak.model.viewable_tag_model.ViewableResourceTag].

    :param v_type: The type of [view][ofrak.resource] to get the children as
    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :param r_sort: Specifies which indexable attribute to use as the key to sort and the
    direction to sort
    :return:

    :raises NotFoundError: If a filter was provided and no resources match the provided filter
    """
    return await self.get_descendants_as_view(v_type, 1, r_filter, r_sort)

get_only_child(self, r_filter=None) async

If a filter is provided, get the only child of this resource which matches the given filter. If a filter is not provided, gets the only child of this resource.

Parameters:

Name Type Description Default
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None

Returns:

Type Description
Resource

Exceptions:

Type Description
NotFoundError

If a filter is provided and more or fewer than one child matches r_filter

NotFoundError

If a filter is not provided and this resource has multiple children

Source code in ofrak/resource.py
async def get_only_child(self, r_filter: ResourceFilter = None) -> "Resource":
    """
    If a filter is provided, get the only child of this resource which matches the given
    filter. If a filter is not provided, gets the only child of this resource.

    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If a filter is provided and more or fewer than one child matches
    ``r_filter``
    :raises NotFoundError: If a filter is not provided and this resource has multiple children
    """
    return await self.get_only_descendant(1, r_filter)

get_only_child_as_view(self, v_type, r_filter=None) async

If a filter is provided, get the only child of this resource which matches the given filter. If a filter is not provided, gets the only child of this resource. The child will be returned as an instance of the given viewable tag.

Parameters:

Name Type Description Default
v_type Type[~RV]

The type of view to get the child as

required
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None

Returns:

Type Description
~RV

Exceptions:

Type Description
NotFoundError

If a filter is provided and more or fewer than one child matches r_filter

NotFoundError

If a filter is not provided and this resource has multiple children

Source code in ofrak/resource.py
async def get_only_child_as_view(self, v_type: Type[RV], r_filter: ResourceFilter = None) -> RV:
    """
    If a filter is provided, get the only child of this resource which matches the given
    filter. If a filter is not provided, gets the only child of this resource. The child will
    be returned as an instance of the given
    [viewable tag][ofrak.model.viewable_tag_model.ViewableResourceTag].

    :param v_type: The type of [view][ofrak.resource] to get the child as
    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :return:

    :raises NotFoundError: If a filter is provided and more or fewer than one child matches
    ``r_filter``
    :raises NotFoundError: If a filter is not provided and this resource has multiple children
    """
    return await self.get_only_descendant_as_view(v_type, 1, r_filter)

delete(self) async

Delete this resource and all of its descendants.

Returns:

Type Description
Source code in ofrak/resource.py
async def delete(self):
    """
    Delete this resource and all of its descendants.

    :return:
    """
    self._component_context.resources_deleted.add(self._resource.id)

    for child_r in await self.get_children():
        await child_r.delete()

    self._resource.is_modified = True
    self._resource.is_deleted = True

flush_data_to_disk(self, path, pack=True) async

Recursively repack the resource and write its data out to a file on disk. If this is a dataless resource, creates an empty file.

Parameters:

Name Type Description Default
path str

Path to the file to write out to. The file is created if it does not exist.

required
Source code in ofrak/resource.py
async def flush_data_to_disk(self, path: str, pack: bool = True):
    """
    Recursively repack the resource and write its data out to a file on disk. If this is a
    dataless resource, creates an empty file.

    :param path: Path to the file to write out to. The file is created if it does not exist.
    """
    if pack is True:
        await self.pack_recursively()

    data = await self.get_data()
    if data is not None:
        with open(path, "wb") as f:
            f.write(data)
    else:
        # Create empty file
        with open(path, "wb") as f:
            pass

summarize(self) async

Create a string summary of this resource, including specific tags, attribute types, and the data offsets of this resource in the parent and root (if applicable).

Not that this is not a complete string representation of the resource: not all tags are included, and only the types of attributes are included, not their values. It is a summary which gives a high level overview of the resource.

Source code in ofrak/resource.py
async def summarize(self) -> str:
    """
    Create a string summary of this resource, including specific tags, attribute types,
    and the data offsets of this resource in the parent and root (if applicable).

    Not that this is not a complete string representation of the resource: not all tags are
    included, and only the types of attributes are included, not their values. It is a
    summary which gives a high level overview of the resource.
    """
    return await _default_summarize_resource(self)

summarize_tree(self, r_filter=None, r_sort=None, indent='', summarize_resource_callback=None) async

Create a string summary of this resource and its (optionally filtered and/or sorted) descendants. The summaries of each resource are the same as the result of summarize, organized into a tree structure. If a filter parameter is provided, it is applied recursively: the children of this resource will be filtered, then only those children matching the filter be displayed, and then the same filter will be applied to their children, etc. For example,

Parameters:

Name Type Description Default
r_filter ResourceFilter

Contains parameters which resources must match to be returned, including any tags it must have and/or values of indexable attributes

None
r_sort ResourceSort

Specifies which indexable attribute to use as the key to sort and the direction to sort

None
Source code in ofrak/resource.py
async def summarize_tree(
    self,
    r_filter: ResourceFilter = None,
    r_sort: ResourceSort = None,
    indent: str = "",
    summarize_resource_callback: Optional[Callable[["Resource"], Awaitable[str]]] = None,
) -> str:
    """
    Create a string summary of this resource and its (optionally filtered and/or sorted)
    descendants. The summaries of each resource are the same as the result of
    [summarize][ofrak.resource.Resource.summarize], organized into a tree structure.
    If a filter parameter is provided, it is applied recursively: the children of this
    resource will be filtered, then only those children matching
    the filter be displayed, and then the same filter will be applied to their children,
    etc. For example,

    :param r_filter: Contains parameters which resources must match to be returned, including
    any tags it must have and/or values of indexable attributes
    :param r_sort: Specifies which indexable attribute to use as the key to sort and the
    direction to sort
    """
    SPACER_BLANK = "   "
    SPACER_LINE = "───"

    if summarize_resource_callback is None:
        summarize_resource_callback = _default_summarize_resource

    children = cast(
        List[Resource], list(await self.get_children(r_filter=r_filter, r_sort=r_sort))
    )

    if children:
        if indent == "":
            tree_string = "┌"
        else:
            tree_string = "┬"
    else:
        tree_string = "─"

    tree_string += f"{await summarize_resource_callback(self)}\n"

    # All children but the last should display as a "fork" in the drop-down tree
    # After the last child, a vertical line should not be drawn as part of the indent
    # Both of those needs are handled here
    child_formatting: List[Tuple[str, str]] = [
        ("├", indent + "│" + SPACER_BLANK) for _ in children[:-1]
    ]
    child_formatting.append(("└", indent + " " + SPACER_BLANK))

    for child, (branch_symbol, child_indent) in zip(children, child_formatting):
        child_tree_string = await child.summarize_tree(
            r_filter=r_filter,
            r_sort=r_sort,
            indent=child_indent,
            summarize_resource_callback=summarize_resource_callback,
        )
        tree_string += f"{indent}{branch_symbol}{SPACER_LINE}{child_tree_string}"
    return tree_string

ResourceFactory

Factory for creating Resource.

create(self, job_id, resource_id, resource_context, resource_view_context, component_context, job_context=None) async

Create a resource from a resource_id.

Parameters:

Name Type Description Default
job_id bytes required
resource_id bytes required
resource_context ResourceContext required
resource_view_context ResourceViewContext required
component_context ComponentContext required
job_context Optional[ofrak.model.job_model.JobRunContext] None
Source code in ofrak/resource.py
async def create(
    self,
    job_id: bytes,
    resource_id: bytes,
    resource_context: ResourceContext,
    resource_view_context: ResourceViewContext,
    component_context: ComponentContext,
    job_context: Optional[JobRunContext] = None,
) -> Resource:
    """
    Create a resource from a resource_id.

    :param job_id:
    :param resource_id:
    :param resource_context:
    :param resource_view_context:
    :param component_context:
    :param job_context:
    """
    resource_m = resource_context.resource_models.get(resource_id)
    if resource_m is None:
        resource_m = MutableResourceModel.from_model(
            await self._resource_service.get_by_id(resource_id)
        )
        resource_context.resource_models[resource_id] = resource_m

    return next(
        iter(
            self._create(
                job_id,
                [resource_m],
                resource_context,
                resource_view_context,
                component_context,
                job_context,
            )
        )
    )

create_many(self, job_id, resource_ids, resource_context, resource_view_context, component_context, job_context=None) async

Create Resources from resource_ids.

Parameters:

Name Type Description Default
job_id bytes required
resource_ids Iterable[bytes] required
resource_context ResourceContext required
resource_view_context ResourceViewContext required
component_context ComponentContext required
job_context Optional[ofrak.model.job_model.JobRunContext] None
Source code in ofrak/resource.py
async def create_many(
    self,
    job_id: bytes,
    resource_ids: Iterable[bytes],
    resource_context: ResourceContext,
    resource_view_context: ResourceViewContext,
    component_context: ComponentContext,
    job_context: Optional[JobRunContext] = None,
) -> Iterable[Resource]:
    """
    Create Resources from resource_ids.

    :param job_id:
    :param resource_ids:
    :param resource_context:
    :param resource_view_context:
    :param component_context:
    :param job_context:
    """
    resource_models_minus_missing: List[Union[MutableResourceModel, int]] = []
    missing_ids: List[bytes] = []

    resource_m: MutableResourceModel

    for resource_id in resource_ids:
        resource_m = resource_context.resource_models.get(resource_id)  # type: ignore
        if resource_m is None:
            resource_models_minus_missing.append(len(missing_ids))
            missing_ids.append(resource_id)
        else:
            resource_models_minus_missing.append(resource_m)

    fetched_models: List[ResourceModel] = list(
        await self._resource_service.get_by_ids(missing_ids)
    )

    resource_models = []
    for resource_model_or_idx in resource_models_minus_missing:
        if type(resource_model_or_idx) is int:
            resource_m = MutableResourceModel.from_model(
                fetched_models[cast(int, resource_model_or_idx)]
            )
            resource_models.append(resource_m)
            resource_context.resource_models[resource_m.id] = resource_m
        else:
            resource_models.append(cast(MutableResourceModel, resource_model_or_idx))

    return self._create(
        job_id,
        resource_models,
        resource_context,
        resource_view_context,
        component_context,
        job_context,
    )