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

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).

Source code in ontocast/tool/triple_manager/core.py 
 18
 19
 20
 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
class TripleStoreManager(Tool):
    """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 []

    @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:
        """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

    @abc.abstractmethod
    async def clean(self, dataset: str | None = None) -> None:
        """Clean/flush data from the triple store.

        This method removes data from the triple store. For Fuseki, the optional
        dataset parameter allows cleaning a specific dataset, or all datasets if None.
        For Neo4j and Filesystem, the dataset parameter is ignored.

        Args:
            dataset: Optional dataset name to clean (Fuseki only). If None, cleans
                all data. For other stores, this parameter is ignored.

        Warning: This operation is irreversible and will delete all data.

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

__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 
29
30
31
32
33
34
35
def __init__(self, **kwargs):
    """Initialize the triple store manager.

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

clean(dataset=None) abstractmethod async

Clean/flush data from the triple store.

This method removes data from the triple store. For Fuseki, the optional dataset parameter allows cleaning a specific dataset, or all datasets if None. For Neo4j and Filesystem, the dataset parameter is ignored.

Parameters:

Name Type Description Default
dataset str | None

Optional dataset name to clean (Fuseki only). If None, cleans all data. For other stores, this parameter is ignored.

 None

Raises:

Type Description
NotImplementedError

If the triple store doesn't support cleaning.
Source code in ontocast/tool/triple_manager/core.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@abc.abstractmethod
async def clean(self, dataset: str | None = None) -> None:
    """Clean/flush data from the triple store.

    This method removes data from the triple store. For Fuseki, the optional
    dataset parameter allows cleaning a specific dataset, or all datasets if None.
    For Neo4j and Filesystem, the dataset parameter is ignored.

    Args:
        dataset: Optional dataset name to clean (Fuseki only). If None, cleans
            all data. For other stores, this parameter is ignored.

    Warning: This operation is irreversible and will delete all data.

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

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 
37
38
39
40
41
42
43
44
45
46
47
@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 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@abc.abstractmethod
def serialize(self, o: Ontology | RDFGraph, **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:
        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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@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

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 
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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)