Skip to content

ontocast.tool.triple_manager

Triple store management tools for OntoCast.

This module provides functionality for managing RDF triple stores, including abstract interfaces and filesystem-based implementations.

FilesystemTripleStoreManager

Bases: TripleStoreManager

Filesystem-based implementation of triple store management.

This class provides a concrete implementation of triple store management using the local filesystem for storage.

Attributes:

Name Type Description
working_directory Path

Path to the working directory for storing data.
ontology_path Optional[Path]

Optional path to the ontology directory.
Source code in ontocast/tool/triple_manager.py 
 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
class FilesystemTripleStoreManager(TripleStoreManager):
    """Filesystem-based implementation of triple store management.

    This class provides a concrete implementation of triple store management
    using the local filesystem for storage.

    Attributes:
        working_directory: Path to the working directory for storing data.
        ontology_path: Optional path to the ontology directory.
    """

    working_directory: pathlib.Path
    ontology_path: Optional[pathlib.Path]

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

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

    def fetch_ontologies(self) -> list[Ontology]:
        """Fetch all available ontologies from the filesystem.

        Returns:
            list[Ontology]: List of available ontologies.
        """
        ontologies = []
        if self.ontology_path is not None:
            sorted_files = sorted(self.ontology_path.glob("*.ttl"))
            for fname in sorted_files:
                try:
                    ontology = Ontology.from_file(fname)
                    ontologies.append(ontology)
                except Exception as e:
                    logging.error(f"Failed to load ontology {fname}: {str(e)}")
        return ontologies

    def serialize_ontology(self, o: Ontology, **kwargs):
        """Store an ontology in the filesystem.

        Args:
            o: The ontology to store.
            **kwargs: Additional keyword arguments for serialization.
        """
        fname = f"ontology_{o.short_name}_{o.version}"
        o.graph.serialize(
            format="turtle", destination=self.working_directory / f"{fname}.ttl"
        )

    def serialize_facts(self, g: Graph, **kwargs):
        """Store a graph in the filesystem.

        Args:
            g: The graph to store.
            **kwargs: Additional keyword arguments for serialization.
                spec: Optional specification for the filename.
        """
        spec = kwargs.pop("spec", None)
        if spec is None:
            fname = "current.ttl"
        elif isinstance(spec, str):
            s = spec.split("/")[-2:]
            s = "_".join([x for x in s if x])
            fname = f"facts_{s}.ttl"
        else:
            raise TypeError(f"string expected for spec {spec}")
        filename = self.working_directory / fname
        g.serialize(format="turtle", destination=filename)

__init__(**kwargs)

Initialize the filesystem triple store manager.

Parameters:

Name Type Description Default
**kwargs

Additional keyword arguments passed to the parent class.

 {}
Source code in ontocast/tool/triple_manager.py 
81
82
83
84
85
86
87
def __init__(self, **kwargs):
    """Initialize the filesystem triple store manager.

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

fetch_ontologies()

Fetch all available ontologies from the filesystem.

Returns:

Type Description
list[Ontology]

list[Ontology]: List of available ontologies.
Source code in ontocast/tool/triple_manager.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def fetch_ontologies(self) -> list[Ontology]:
    """Fetch all available ontologies from the filesystem.

    Returns:
        list[Ontology]: List of available ontologies.
    """
    ontologies = []
    if self.ontology_path is not None:
        sorted_files = sorted(self.ontology_path.glob("*.ttl"))
        for fname in sorted_files:
            try:
                ontology = Ontology.from_file(fname)
                ontologies.append(ontology)
            except Exception as e:
                logging.error(f"Failed to load ontology {fname}: {str(e)}")
    return ontologies

serialize_facts(g, **kwargs)

Store a graph in the filesystem.

Parameters:

Name Type Description Default
g Graph

The graph to store.

 required
**kwargs

Additional keyword arguments for serialization. spec: Optional specification for the filename.

 {}
Source code in ontocast/tool/triple_manager.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def serialize_facts(self, g: Graph, **kwargs):
    """Store a graph in the filesystem.

    Args:
        g: The graph to store.
        **kwargs: Additional keyword arguments for serialization.
            spec: Optional specification for the filename.
    """
    spec = kwargs.pop("spec", None)
    if spec is None:
        fname = "current.ttl"
    elif isinstance(spec, str):
        s = spec.split("/")[-2:]
        s = "_".join([x for x in s if x])
        fname = f"facts_{s}.ttl"
    else:
        raise TypeError(f"string expected for spec {spec}")
    filename = self.working_directory / fname
    g.serialize(format="turtle", destination=filename)

serialize_ontology(o, **kwargs)

Store an ontology in the filesystem.

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.py 
106
107
108
109
110
111
112
113
114
115
116
def serialize_ontology(self, o: Ontology, **kwargs):
    """Store an ontology in the filesystem.

    Args:
        o: The ontology to store.
        **kwargs: Additional keyword arguments for serialization.
    """
    fname = f"ontology_{o.short_name}_{o.version}"
    o.graph.serialize(
        format="turtle", destination=self.working_directory / f"{fname}.ttl"
    )

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.

Source code in ontocast/tool/triple_manager.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
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.

    """

    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.

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

    @abc.abstractmethod
    def serialize_ontology(self, o: Ontology, **kwargs):
        """Store an ontology in the triple store.

        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 a given name.

        Args:
            g: The graph 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.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.

Returns:

Type Description
list[Ontology]

list[Ontology]: List of available ontologies.
Source code in ontocast/tool/triple_manager.py 
37
38
39
40
41
42
43
44
@abc.abstractmethod
def fetch_ontologies(self) -> list[Ontology]:
    """Fetch all available ontologies.

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

serialize_facts(g, **kwargs) abstractmethod

Store a graph with a given name.

Parameters:

Name Type Description Default
g Graph

The graph to store.

 required
**kwargs

Additional keyword arguments for serialization.

 {}
Source code in ontocast/tool/triple_manager.py 
56
57
58
59
60
61
62
63
64
@abc.abstractmethod
def serialize_facts(self, g: Graph, **kwargs):
    """Store a graph with a given name.

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

serialize_ontology(o, **kwargs) abstractmethod

Store an ontology in the triple store.

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.py 
46
47
48
49
50
51
52
53
54
@abc.abstractmethod
def serialize_ontology(self, o: Ontology, **kwargs):
    """Store an ontology in the triple store.

    Args:
        o: The ontology to store.
        **kwargs: Additional keyword arguments for serialization.
    """
    pass