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
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_ontology(self, o: Ontology, **kwargs):
        """Store an ontology in the triple store.

        This method should store the given ontology and its associated 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: The ontology to store.
            **kwargs: Additional keyword arguments for serialization.
        """
        pass

    @abc.abstractmethod
    def serialize_facts(self, g: Graph, **kwargs):
        """Store a graph with facts in the triple store.

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

        Args:
            g: The RDF graph containing facts to store.
            **kwargs: Additional keyword arguments for serialization.
        """
        pass

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

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_facts(g, **kwargs) abstractmethod

Store a graph with facts in the triple store.

This method should store the given RDF graph containing facts 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
g Graph

The RDF graph containing facts to store.

 required
**kwargs

Additional keyword arguments for serialization.

 {}
Source code in ontocast/tool/triple_manager/core.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
@abc.abstractmethod
def serialize_facts(self, g: Graph, **kwargs):
    """Store a graph with facts in the triple store.

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

    Args:
        g: The RDF graph containing facts to store.
        **kwargs: Additional keyword arguments for serialization.
    """
    pass

serialize_ontology(o, **kwargs) abstractmethod

Store an ontology in the triple store.

This method should store the given ontology and its associated 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

The ontology to store.

 required
**kwargs

Additional keyword arguments for serialization.

 {}
Source code in ontocast/tool/triple_manager/core.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
@abc.abstractmethod
def serialize_ontology(self, o: Ontology, **kwargs):
    """Store an ontology in the triple store.

    This method should store the given ontology and its associated 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: The ontology to store.
        **kwargs: Additional keyword arguments for serialization.
    """
    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 Optional[str]

The connection URI for the triple store.
auth Optional[tuple]

Authentication tuple (username, password) for the triple store.
Source code in ontocast/tool/triple_manager/core.py 
 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
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: Optional[str] = Field(default=None, description="Triple store connection URI")
    auth: Optional[tuple] = Field(
        default=None, description="Triple store authentication tuple (user, password)"
    )
    clean: bool = Field(
        default=False, description="If True, clean the database on init."
    )

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