collection

Backend for DLite collections.

CollectionStrategy

Triplestore strategy for DLite collections.

Parameters:

Name	Type	Description	Default
base_iri	Optional[str]	Unused.	None
database	Optional[str]	Unused - collection does not support multiple databases.	None
collection	Optional[Union[dlite.Collection, str]]	Optional collection from which to initialise the triplestore from.	None

Source code in tripper/backends/collection.py

class CollectionStrategy:
    """Triplestore strategy for DLite collections.

    Arguments:
        base_iri: Unused.
        database: Unused - collection does not support multiple databases.
        collection: Optional collection from which to initialise the
            triplestore from.
    """

    prefer_sparql = False

    def __init__(
        self,
        base_iri: "Optional[str]" = None,
        database: "Optional[str]" = None,
        collection: "Optional[Union[dlite.Collection, str]]" = None,
    ):
        # pylint: disable=unused-argument
        if collection is None:
            self.collection = dlite.Collection()
        elif isinstance(collection, str):
            self.collection = dlite.get_instance(collection)
            if self.collection.meta.uri != dlite.COLLECTION_ENTITY:
                raise TypeError(
                    f"expected '{collection}' to be a collection, was a "
                    f"{self.collection.meta.uri}"
                )
        elif isinstance(collection, dlite.Collection):
            self.collection = collection
        else:
            raise TypeError(
                "`collection` should be None, string or a collection"
            )

    def triples(self, triple: "Triple") -> "Generator[Triple, None, None]":
        """Returns a generator over matching triples."""
        for s, p, o, d in self.collection.get_relations(*triple, rettype="T"):
            if d:
                lang = d[1:] if d[0] == "@" else None
                dt = None if lang else d
                yield s, p, Literal(o, lang=lang, datatype=dt)
            else:
                yield s, p, o

    def add_triples(
        self, triples: "Union[Sequence[Triple], Generator[Triple, None, None]]"
    ):
        """Add a sequence of triples."""
        for s, p, o in triples:
            v = parse_object(o)
            obj = v if isinstance(v, str) else str(v.value)
            d = (
                None
                if not isinstance(v, Literal)
                else f"@{v.lang}" if v.lang else v.datatype
            )
            self.collection.add_relation(s, p, obj, d)

    def remove(self, triple: "Triple"):
        """Remove all matching triples from the backend."""
        s, p, o = triple
        v = parse_object(o)
        obj = v if isinstance(v, str) else str(v.value)
        d = (
            None
            if not isinstance(v, Literal)
            else f"@{v.lang}" if v.lang else v.datatype
        )
        self.collection.remove_relations(s, p, obj, d)

add_triples(self, triples)

Add a sequence of triples.

Source code in tripper/backends/collection.py

def add_triples(
    self, triples: "Union[Sequence[Triple], Generator[Triple, None, None]]"
):
    """Add a sequence of triples."""
    for s, p, o in triples:
        v = parse_object(o)
        obj = v if isinstance(v, str) else str(v.value)
        d = (
            None
            if not isinstance(v, Literal)
            else f"@{v.lang}" if v.lang else v.datatype
        )
        self.collection.add_relation(s, p, obj, d)

remove(self, triple)

Remove all matching triples from the backend.

Source code in tripper/backends/collection.py

def remove(self, triple: "Triple"):
    """Remove all matching triples from the backend."""
    s, p, o = triple
    v = parse_object(o)
    obj = v if isinstance(v, str) else str(v.value)
    d = (
        None
        if not isinstance(v, Literal)
        else f"@{v.lang}" if v.lang else v.datatype
    )
    self.collection.remove_relations(s, p, obj, d)

triples(self, triple)

Returns a generator over matching triples.

Source code in tripper/backends/collection.py

def triples(self, triple: "Triple") -> "Generator[Triple, None, None]":
    """Returns a generator over matching triples."""
    for s, p, o, d in self.collection.get_relations(*triple, rettype="T"):
        if d:
            lang = d[1:] if d[0] == "@" else None
            dt = None if lang else d
            yield s, p, Literal(o, lang=lang, datatype=dt)
        else:
            yield s, p, o