Aspect

An Aspect is a typed, time-bounded assertion attached to any entity URN in the Datafabric. Aspects are the building blocks of IVCAP's knowledge graph.

Quick Reference

from ivcap_client.ivcap import IVCAP

ivcap = IVCAP()

# Add an aspect
aspect = ivcap.add_aspect(
    entity="urn:ivcap:artifact:<uuid>",
    aspect={
        "$schema": "urn:my-project:schema:annotation.1",
        "label": "coral",
        "confidence": 0.97,
    },
)

# List aspects for an entity
for aspect in ivcap.list_aspects(entity="urn:ivcap:artifact:<uuid>"):
    print(aspect.schema, aspect.entity)
    print(aspect.aspect)   # JSON content

# Retract
aspect.retract()

Class Documentation

Aspect dataclass

This class represents a aspect record stored at a particular IVCAP deployment

Source code in ivcap_client/aspect.py

@dataclass
class Aspect:
    """This class represents a aspect record
    stored at a particular IVCAP deployment"""

    id: str
    entity: str
    schema: str

    # content: Optional[any] = None
    content_type: str | None = None

    valid_from: datetime.datetime | None = None
    valid_to: datetime.datetime | None = None

    asserter: URN | None = None
    retracter: URN | None = None

    @classmethod
    def _from_list_item(cls, item: AspectListItemRT, ivcap: IVCAP):
        kwargs = item.to_dict()
        return cls(ivcap, **kwargs)

    def __init__(self, ivcap: IVCAP, **kwargs):
        if not ivcap:
            raise ValueError("missing 'ivcap' argument")
        self._ivcap = ivcap
        self.__update__(**kwargs)

    def __update__(self, **kwargs):
        p = [
            "id",
            "entity",
            "schema",
            "content-type",
            "valid-from",
            "valid-to",
            "asserter",
            "retracter",
        ]
        hp = ["content"]
        _set_fields(self, p, hp, kwargs)

        c = kwargs.get("content")
        if isinstance(c, dict):
            self._content = c
        else:
            self._content = None

    @property
    def urn(self) -> str:
        return self.id

    @property
    def aspect(self) -> dict:
        if self._content is None:
            self.refresh()
        return self._content

    @property
    def content(self) -> dict:
        return self.aspect

    def refresh(self) -> Aspect:
        r = aspect_read.sync_detailed(self.id, client=self._ivcap._client)
        if r.status_code >= 300:
            return process_error("aspect", r)
        res: AspectRT = r.parsed
        self.__update__(**res.to_dict())
        return self

    def retract(self) -> Aspect:
        """Retract this aspect"""
        if self.valid_to:
            # already retracted
            return self
        r = aspect_retract.sync_detailed(self.id, client=self._ivcap._client)
        if r.status_code >= 300:
            return process_error("aspect", r)
        return self.refresh()

    def __repr__(self):
        return f"<Aspect id={self.id}, entity={self.entity} schema={self.schema}>"

retract()

Retract this aspect

Source code in ivcap_client/aspect.py

def retract(self) -> Aspect:
    """Retract this aspect"""
    if self.valid_to:
        # already retracted
        return self
    r = aspect_retract.sync_detailed(self.id, client=self._ivcap._client)
    if r.status_code >= 300:
        return process_error("aspect", r)
    return self.refresh()

Copy Markdown