Skip to content

ontocast.tool.triple_manager.core

Triple store management tools for OntoCast.

This module provides functionality for managing RDF triple stores, including abstract interfaces and concrete implementations for different triple store backends.

TripleStoreManager

Bases: Tool

Source code in ontocast/tool/triple_manager/core.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class TripleStoreManager(Tool):
    _PROVENANCE_METADATA_PREDICATES: ClassVar[set] = {
        PROV.generatedAtTime,
        SCHEMA.position,
        SCHEMA.identifier,
    }

    """Base class for managing RDF triple stores.

    This class defines the interface for triple store management operations,
    including fetching and storing ontologies and their graphs. All concrete
    triple store implementations should inherit from this class.

    This is an abstract base class that must be implemented by specific
    triple store backends (e.g., Neo4j, Fuseki, Filesystem).
    """

    def __init__(self, **kwargs):
        """Initialize the triple store manager.

        Args:
            **kwargs: Additional keyword arguments passed to the parent class.
        """
        super().__init__(**kwargs)

    @abc.abstractmethod
    def fetch_ontologies(self) -> list[Ontology]:
        """Fetch all available ontologies from the triple store.

        This method should retrieve all ontologies stored in the triple store
        and return them as Ontology objects with their associated RDF graphs.

        Returns:
            list[Ontology]: List of available ontologies with their graphs.
        """
        return []

    async def afetch_ontologies(self) -> list[Ontology]:
        """Async fetch helper for backends without native async I/O."""
        return await asyncio.to_thread(self.fetch_ontologies)

    @abc.abstractmethod
    def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
        """Store an RDF graph in the triple store.

        This method should store the given RDF graph in the triple store.
        The implementation may choose how to organize the storage (e.g., as named graphs,
        in specific collections, etc.).

        Args:
            graph: The RDF graph to store.
            **kwargs: Implementation-specific arguments (e.g., fname for filesystem, graph_uri for Fuseki).

        Returns:
            bool | None: Implementation-specific return value (bool for Fuseki, summary for Neo4j, None for Filesystem).
        """
        pass

    @abc.abstractmethod
    def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:  # type: ignore[override]
        """Store an RDF graph in the triple store.

        This method should store the given RDF graph in the triple store.
        The implementation may choose how to organize the storage (e.g., as named graphs,
        in specific collections, etc.).

        Args:
            o: RDF graph or Ontology object to store.
            **kwargs: Implementation-specific arguments (e.g., graph_uri for Fuseki).

        Returns:
            bool | None: Implementation-specific return value (bool for Fuseki, summary for Neo4j, None for Filesystem).
        """
        pass

    async def aserialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
        """Async serialize helper for backends without native async I/O."""
        return await asyncio.to_thread(self.serialize, o, **kwargs)

    @classmethod
    def strip_provenance(cls, graph: Graph) -> RDFGraph:
        """Return a graph without reification/provenance scaffolding triples."""
        clean = RDFGraph()
        for prefix, namespace in graph.namespaces():
            clean.bind(prefix, namespace)

        reifier_nodes = set(graph.subjects(RDF_REIFIES, None))
        source_nodes = set(graph.objects(None, PROV.wasDerivedFrom))

        for subject, predicate, object_ in graph:
            if predicate in {RDF_REIFIES, PROV.wasDerivedFrom}:
                continue
            if subject in reifier_nodes:
                continue
            if subject in source_nodes:
                continue
            clean.add((subject, predicate, object_))

        return clean

    @abc.abstractmethod
    async def clean(self) -> None:
        """Clean/flush data managed by this store (backend-specific scope).

        Warning: This operation is irreversible and will delete data.

        Raises:
            NotImplementedError: If the triple store doesn't support cleaning.
        """
        raise NotImplementedError("clean() method must be implemented by subclasses")

    def supports_tenancy_partition(self) -> bool:
        """True if this backend isolates facts/ontologies by :func:`tenant_project_*` names."""
        return False

    async def clean_tenancy(self, tenant: str, project: str) -> None:
        """Remove all triples for datasets derived from ``tenant`` / ``project``.

        Backends without per-tenant partitions raise :class:`NotImplementedError`.
        """
        raise NotImplementedError(
            f"{type(self).__name__} does not isolate data by tenant/project"
        )

__init__(**kwargs)

Initialize the triple store manager.

Parameters:

Name Type Description Default
**kwargs

Additional keyword arguments passed to the parent class.

 {}
Source code in ontocast/tool/triple_manager/core.py 
38
39
40
41
42
43
44
def __init__(self, **kwargs):
    """Initialize the triple store manager.

    Args:
        **kwargs: Additional keyword arguments passed to the parent class.
    """
    super().__init__(**kwargs)

afetch_ontologies() async

Async fetch helper for backends without native async I/O.

Source code in ontocast/tool/triple_manager/core.py 
58
59
60
async def afetch_ontologies(self) -> list[Ontology]:
    """Async fetch helper for backends without native async I/O."""
    return await asyncio.to_thread(self.fetch_ontologies)

aserialize(o, **kwargs) async

Async serialize helper for backends without native async I/O.

Source code in ontocast/tool/triple_manager/core.py 
96
97
98
async def aserialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
    """Async serialize helper for backends without native async I/O."""
    return await asyncio.to_thread(self.serialize, o, **kwargs)

clean() abstractmethod async

Clean/flush data managed by this store (backend-specific scope).

Warning: This operation is irreversible and will delete data.

Raises:

Type Description
NotImplementedError

If the triple store doesn't support cleaning.
Source code in ontocast/tool/triple_manager/core.py 
121
122
123
124
125
126
127
128
129
130
@abc.abstractmethod
async def clean(self) -> None:
    """Clean/flush data managed by this store (backend-specific scope).

    Warning: This operation is irreversible and will delete data.

    Raises:
        NotImplementedError: If the triple store doesn't support cleaning.
    """
    raise NotImplementedError("clean() method must be implemented by subclasses")

clean_tenancy(tenant, project) async

Remove all triples for datasets derived from tenant / project.

Backends without per-tenant partitions raise :class:NotImplementedError.

Source code in ontocast/tool/triple_manager/core.py 
136
137
138
139
140
141
142
143
async def clean_tenancy(self, tenant: str, project: str) -> None:
    """Remove all triples for datasets derived from ``tenant`` / ``project``.

    Backends without per-tenant partitions raise :class:`NotImplementedError`.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not isolate data by tenant/project"
    )

fetch_ontologies() abstractmethod

Fetch all available ontologies from the triple store.

This method should retrieve all ontologies stored in the triple store and return them as Ontology objects with their associated RDF graphs.

Returns:

Type Description
list[Ontology]

list[Ontology]: List of available ontologies with their graphs.
Source code in ontocast/tool/triple_manager/core.py 
46
47
48
49
50
51
52
53
54
55
56
@abc.abstractmethod
def fetch_ontologies(self) -> list[Ontology]:
    """Fetch all available ontologies from the triple store.

    This method should retrieve all ontologies stored in the triple store
    and return them as Ontology objects with their associated RDF graphs.

    Returns:
        list[Ontology]: List of available ontologies with their graphs.
    """
    return []

serialize(o, **kwargs) abstractmethod

Store an RDF graph in the triple store.

This method should store the given RDF graph in the triple store. The implementation may choose how to organize the storage (e.g., as named graphs, in specific collections, etc.).

Parameters:

Name Type Description Default
o Ontology | RDFGraph

RDF graph or Ontology object to store.

 required
**kwargs

Implementation-specific arguments (e.g., graph_uri for Fuseki).

 {}

Returns:

Type Description
bool | None

bool | None: Implementation-specific return value (bool for Fuseki, summary for Neo4j, None for Filesystem).
Source code in ontocast/tool/triple_manager/core.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@abc.abstractmethod
def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:  # type: ignore[override]
    """Store an RDF graph in the triple store.

    This method should store the given RDF graph in the triple store.
    The implementation may choose how to organize the storage (e.g., as named graphs,
    in specific collections, etc.).

    Args:
        o: RDF graph or Ontology object to store.
        **kwargs: Implementation-specific arguments (e.g., graph_uri for Fuseki).

    Returns:
        bool | None: Implementation-specific return value (bool for Fuseki, summary for Neo4j, None for Filesystem).
    """
    pass

serialize_graph(graph, **kwargs) abstractmethod

Store an RDF graph in the triple store.

This method should store the given RDF graph in the triple store. The implementation may choose how to organize the storage (e.g., as named graphs, in specific collections, etc.).

Parameters:

Name Type Description Default
graph Graph

The RDF graph to store.

 required
**kwargs

Implementation-specific arguments (e.g., fname for filesystem, graph_uri for Fuseki).

 {}

Returns:

Type Description
bool | None

bool | None: Implementation-specific return value (bool for Fuseki, summary for Neo4j, None for Filesystem).
Source code in ontocast/tool/triple_manager/core.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@abc.abstractmethod
def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
    """Store an RDF graph in the triple store.

    This method should store the given RDF graph in the triple store.
    The implementation may choose how to organize the storage (e.g., as named graphs,
    in specific collections, etc.).

    Args:
        graph: The RDF graph to store.
        **kwargs: Implementation-specific arguments (e.g., fname for filesystem, graph_uri for Fuseki).

    Returns:
        bool | None: Implementation-specific return value (bool for Fuseki, summary for Neo4j, None for Filesystem).
    """
    pass

strip_provenance(graph) classmethod

Return a graph without reification/provenance scaffolding triples.

Source code in ontocast/tool/triple_manager/core.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@classmethod
def strip_provenance(cls, graph: Graph) -> RDFGraph:
    """Return a graph without reification/provenance scaffolding triples."""
    clean = RDFGraph()
    for prefix, namespace in graph.namespaces():
        clean.bind(prefix, namespace)

    reifier_nodes = set(graph.subjects(RDF_REIFIES, None))
    source_nodes = set(graph.objects(None, PROV.wasDerivedFrom))

    for subject, predicate, object_ in graph:
        if predicate in {RDF_REIFIES, PROV.wasDerivedFrom}:
            continue
        if subject in reifier_nodes:
            continue
        if subject in source_nodes:
            continue
        clean.add((subject, predicate, object_))

    return clean

supports_tenancy_partition()

True if this backend isolates facts/ontologies by :func:tenant_project_* names.

Source code in ontocast/tool/triple_manager/core.py 
132
133
134
def supports_tenancy_partition(self) -> bool:
    """True if this backend isolates facts/ontologies by :func:`tenant_project_*` names."""
    return False

TripleStoreManagerWithAuth

Bases: TripleStoreManager

Base class for triple store managers that require authentication.

This class provides common functionality for triple store managers that need URI and authentication credentials. It handles environment variable loading and credential parsing.

Attributes:

Name Type Description
uri str | None

The connection URI for the triple store.
auth tuple | None

Authentication tuple (username, password) for the triple store.
Source code in ontocast/tool/triple_manager/core.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class TripleStoreManagerWithAuth(TripleStoreManager):
    """Base class for triple store managers that require authentication.

    This class provides common functionality for triple store managers that
    need URI and authentication credentials. It handles environment variable
    loading and credential parsing.

    Attributes:
        uri: The connection URI for the triple store.
        auth: Authentication tuple (username, password) for the triple store.
    """

    uri: str | None = Field(default=None, description="Triple store connection URI")
    auth: tuple | None = Field(
        default=None, description="Triple store authentication tuple (user, password)"
    )

    def __init__(self, uri=None, auth=None, env_uri=None, env_auth=None, **kwargs):
        """Initialize the triple store manager with authentication.

        This method handles loading URI and authentication credentials from
        either direct parameters or environment variables. It also parses
        authentication strings in the format "user/password".

        Args:
            uri: Direct URI for the triple store connection.
            auth: Direct authentication tuple or string in "user/password" format.
            env_uri: Environment variable name for the URI (e.g., "NEO4J_URI").
            env_auth: Environment variable name for authentication (e.g., "NEO4J_AUTH").
            **kwargs: Additional keyword arguments passed to the parent class.

        Raises:
            ValueError: If authentication string is not in "user/password" format.

        Example:
            >>> manager = TripleStoreManagerWithAuth(
            ...     env_uri="NEO4J_URI",
            ...     env_auth="NEO4J_AUTH"
            ... )
        """
        # Use env vars if not provided
        uri = uri or (os.getenv(env_uri) if env_uri else None)
        auth_env = auth or (os.getenv(env_auth) if env_auth else None)

        if auth_env and not isinstance(auth_env, tuple):
            if "/" in auth_env:
                user, password = auth_env.split("/", 1)
                auth = (user, password)
            else:
                raise ValueError(
                    f"{env_auth or 'TRIPLESTORE_AUTH'} must be in 'user/password' format"
                )
        elif isinstance(auth_env, tuple):
            auth = auth_env
        # else: auth remains None

        super().__init__(uri=uri, auth=auth, **kwargs)

__init__(uri=None, auth=None, env_uri=None, env_auth=None, **kwargs)

Initialize the triple store manager with authentication.

This method handles loading URI and authentication credentials from either direct parameters or environment variables. It also parses authentication strings in the format "user/password".

Parameters:

Name Type Description Default
uri

Direct URI for the triple store connection.

 None
auth

Direct authentication tuple or string in "user/password" format.

 None
env_uri

Environment variable name for the URI (e.g., "NEO4J_URI").

 None
env_auth

Environment variable name for authentication (e.g., "NEO4J_AUTH").

 None
**kwargs

Additional keyword arguments passed to the parent class.

 {}

Raises:

Type Description
ValueError

If authentication string is not in "user/password" format.
Example

manager = TripleStoreManagerWithAuth( ... env_uri="NEO4J_URI", ... env_auth="NEO4J_AUTH" ... )

Source code in ontocast/tool/triple_manager/core.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def __init__(self, uri=None, auth=None, env_uri=None, env_auth=None, **kwargs):
    """Initialize the triple store manager with authentication.

    This method handles loading URI and authentication credentials from
    either direct parameters or environment variables. It also parses
    authentication strings in the format "user/password".

    Args:
        uri: Direct URI for the triple store connection.
        auth: Direct authentication tuple or string in "user/password" format.
        env_uri: Environment variable name for the URI (e.g., "NEO4J_URI").
        env_auth: Environment variable name for authentication (e.g., "NEO4J_AUTH").
        **kwargs: Additional keyword arguments passed to the parent class.

    Raises:
        ValueError: If authentication string is not in "user/password" format.

    Example:
        >>> manager = TripleStoreManagerWithAuth(
        ...     env_uri="NEO4J_URI",
        ...     env_auth="NEO4J_AUTH"
        ... )
    """
    # Use env vars if not provided
    uri = uri or (os.getenv(env_uri) if env_uri else None)
    auth_env = auth or (os.getenv(env_auth) if env_auth else None)

    if auth_env and not isinstance(auth_env, tuple):
        if "/" in auth_env:
            user, password = auth_env.split("/", 1)
            auth = (user, password)
        else:
            raise ValueError(
                f"{env_auth or 'TRIPLESTORE_AUTH'} must be in 'user/password' format"
            )
    elif isinstance(auth_env, tuple):
        auth = auth_env
    # else: auth remains None

    super().__init__(uri=uri, auth=auth, **kwargs)