Skip to content

ontocast.tool

Tool package for OntoCast.

This package provides a collection of tools that support the OntoCast workflow, including document processing, ontology management, triple store operations, and LLM interactions.

The package includes: - LLMTool: Language model interaction and prompting - OntologyManager: Ontology loading and management - TripleStoreManager: Abstract interface for triple store operations - FusekiTripleStoreManager: Fuseki-specific triple store implementation (preferred) - Neo4jTripleStoreManager: Neo4j-specific triple store implementation - FilesystemTripleStoreManager: Filesystem-based triple store implementation - ConverterTool: Document format conversion utilities - ChunkerTool: Text chunking and segmentation

All tools inherit from the base Tool class and provide standardized interfaces for integration into the OntoCast workflow.

Example

from ontocast.tool import LLMTool, OntologyManager llm = LLMTool.create(provider="openai", model="gpt-4") om = OntologyManager()

ChunkerTool

Bases: Tool

Tool for semantic chunking of documents.

Falls back to naive chunking if sentence-transformers is not available. Includes caching to avoid re-chunking the same text with the same parameters.

Source code in ontocast/tool/chunk/chunker.py 
 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
144
145
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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
class ChunkerTool(Tool):
    """Tool for semantic chunking of documents.

    Falls back to naive chunking if sentence-transformers is not available.
    Includes caching to avoid re-chunking the same text with the same parameters.
    """

    model: str = Field(
        default="sentence-transformers/paraphrase-multilingual-mpnet-base-v2",
        description="HuggingFace model name for embeddings",
    )
    config: ChunkConfig = Field(
        default_factory=ChunkConfig, description="Chunking configuration parameters"
    )
    chunking_mode: Literal["semantic", "naive"] = Field(
        default="semantic" if SEMANTIC_CHUNKING_AVAILABLE else "naive",
        description="Chunking mode: semantic (requires sentence-transformers) or naive (fallback)",
    )
    cache: Any = Field(default=None, exclude=True)

    def __init__(
        self,
        chunk_config: ChunkConfig | None = None,
        cache: Cacher | None = None,
        **kwargs,
    ):
        """Initialize the ChunkerTool.

        Args:
            chunk_config: Chunking configuration. If None, uses default ChunkConfig.
            cache: Optional shared Cacher instance. If None, creates a new one.
            **kwargs: Additional keyword arguments passed to the parent class.
        """
        super().__init__(**kwargs)
        self._model = None
        self._model_lock = threading.Lock()  # Lock for thread-safe model initialization

        # Initialize cache - use shared cacher or create new one
        if cache is not None:
            self.cache = ToolCacher(cache, "chunker")
        else:
            # Fallback for backward compatibility
            shared_cache = Cacher()
            self.cache = ToolCacher(shared_cache, "chunker")

        # Override config if provided
        if chunk_config is not None:
            self.config = chunk_config

        # Override chunking mode if semantic chunking is not available
        if not SEMANTIC_CHUNKING_AVAILABLE and self.chunking_mode == "semantic":
            self.chunking_mode = "naive"
            logger.warning(
                "Semantic chunking not available (sentence-transformers not installed). "
                "Falling back to naive chunking."
            )

    def _init_model(self):
        """Initialize the embedding model in a thread-safe manner.

        Uses double-checked locking pattern to ensure the model is only
        initialized once, even when called concurrently from multiple threads.
        """
        # Fast path: if model already initialized, return immediately
        if self._model is not None:
            return

        # Acquire lock for thread-safe initialization
        with self._model_lock:
            # Double-check: another thread might have initialized it while we waited
            if self._model is None and SEMANTIC_CHUNKING_AVAILABLE:
                if HuggingFaceEmbeddings is not None:  # type: ignore
                    try:
                        self._model = HuggingFaceEmbeddings(
                            model_name=self.model,
                            model_kwargs={
                                "device": "cuda"
                                if torch is not None and torch.cuda.is_available()
                                else "cpu"
                            },
                            encode_kwargs={"normalize_embeddings": False},
                        )
                        logger.debug(f"Initialized embedding model: {self.model}")
                    except Exception as e:
                        logger.error(f"Failed to initialize embedding model: {e}")
                        # Set to a sentinel value to prevent repeated failed attempts
                        self._model = False  # type: ignore

    def _naive_chunk(self, doc: str) -> list[str]:
        """Naive chunking fallback when semantic chunking is not available.

                Args:
                    doc: The document text to chunk.
        git
                Returns:
                    List of text chunks.
        """
        # Split by paragraphs first (double newlines)
        paragraphs = re.split(r"\n\s*\n", doc.strip())

        chunks = []
        current_chunk = ""

        for paragraph in paragraphs:
            paragraph = paragraph.strip()
            if not paragraph:
                continue

            # If adding this paragraph would exceed max_size, start a new chunk
            if (
                current_chunk
                and len(current_chunk) + len(paragraph) + 2 > self.config.max_size
            ):
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = paragraph
            else:
                if current_chunk:
                    current_chunk += "\n\n" + paragraph
                else:
                    current_chunk = paragraph

            # If a single paragraph is too large, split it by sentences
            if len(current_chunk) > self.config.max_size:
                # Save the previous chunk if it exists
                if len(current_chunk) - len(paragraph) - 2 > 0:
                    prev_chunk = current_chunk[
                        : len(current_chunk) - len(paragraph) - 2
                    ].strip()
                    if prev_chunk:
                        chunks.append(prev_chunk)

                # Split the large paragraph by sentences
                sentences = re.split(r"(?<=[.!?])\s+", paragraph)
                temp_chunk = ""

                for sentence in sentences:
                    if len(temp_chunk) + len(sentence) + 1 > self.config.max_size:
                        if temp_chunk:
                            chunks.append(temp_chunk.strip())
                        temp_chunk = sentence
                    else:
                        if temp_chunk:
                            temp_chunk += " " + sentence
                        else:
                            temp_chunk = sentence

                current_chunk = temp_chunk

        # Add the last chunk
        if current_chunk:
            chunks.append(current_chunk.strip())

        # Filter out chunks that are too small
        chunks = [chunk for chunk in chunks if len(chunk) >= self.config.min_size]

        logger.info(f"Naive chunking produced {len(chunks)} chunks")
        return chunks

    def __call__(self, doc: str) -> list[str]:
        """Chunk the document using either semantic or naive chunking.

        Args:
            doc: The document text to chunk.

        Returns:
            List of text chunks.
        """
        # Prepare configuration for caching
        config_dict = {
            "model": self.model,
            "chunking_mode": self.chunking_mode,
            "max_size": self.config.max_size,
            "min_size": self.config.min_size,
            "buffer_size": self.config.buffer_size,
            "breakpoint_threshold_type": self.config.breakpoint_threshold_type,
            "breakpoint_threshold_amount": self.config.breakpoint_threshold_amount,
        }

        # Check cache first
        cached_result = self.cache.get(doc, config=config_dict)
        if cached_result is not None:
            logger.debug("Cache hit for document chunking")
            return cached_result

        # Perform chunking
        if self.chunking_mode == "naive":
            result = self._naive_chunk(doc)
        else:
            # Semantic chunking (requires sentence-transformers)
            if not SEMANTIC_CHUNKING_AVAILABLE:
                logger.warning(
                    "Semantic chunking requested but not available. Falling back to naive chunking."
                )
                result = self._naive_chunk(doc)
            else:
                self._init_model()
                documents = [doc]

                if self._model is None or self._model is False:
                    logger.warning(
                        "Model not initialized. Falling back to naive chunking."
                    )
                    result = self._naive_chunk(doc)
                elif SemanticChunker is None:  # type: ignore
                    logger.warning(
                        "SemanticChunker not available. Falling back to naive chunking."
                    )
                    result = self._naive_chunk(doc)
                else:
                    text_splitter = SemanticChunker(
                        buffer_size=self.config.buffer_size,
                        breakpoint_threshold_type=self.config.breakpoint_threshold_type,
                        breakpoint_threshold_amount=self.config.breakpoint_threshold_amount,
                        embeddings=self._model,
                        min_chunk_size=self.config.min_size,
                        sentence_split_regex=r"(?:(?:\n{2,}(?=#+))|(?:\n{2,}(?=- ))"
                        r"|(?<=[a-z][.?!])\s+(?=\b[A-Z]\w{8,}\b)|(?<!#)(?=#+))",
                    )

                    def recursive_chunking(docs_list: list[str], stop_flag=False):
                        lens = [len(d) for d in docs_list]
                        logger.info(f"chunk lengths: {lens}")
                        if (
                            all(len(d) < self.config.max_size for d in docs_list)
                            or stop_flag
                        ):
                            return docs_list
                        else:
                            new_docs = []
                            for d in docs_list:
                                if len(d) > self.config.max_size:
                                    cdocs_ = text_splitter.create_documents([d])
                                    cdocs = [d.page_content for d in cdocs_]
                                    if len(cdocs[-1]) < self.config.min_size:
                                        cdocs = cdocs[:-2] + [cdocs[-2] + cdocs[-1]]
                                    new_docs.extend(cdocs)
                                else:
                                    new_docs.append(d)
                            stop_flag = len(docs_list) == len(new_docs)
                            return recursive_chunking(new_docs, stop_flag=stop_flag)

                    result = recursive_chunking(documents)

        # Cache the result
        self.cache.set(doc, result, config=config_dict)
        logger.debug("Cached document chunking result")

        return result

__call__(doc)

Chunk the document using either semantic or naive chunking.

Parameters:

Name Type Description Default
doc str

The document text to chunk.

 required

Returns:

Type Description
list[str]

List of text chunks.
Source code in ontocast/tool/chunk/chunker.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def __call__(self, doc: str) -> list[str]:
    """Chunk the document using either semantic or naive chunking.

    Args:
        doc: The document text to chunk.

    Returns:
        List of text chunks.
    """
    # Prepare configuration for caching
    config_dict = {
        "model": self.model,
        "chunking_mode": self.chunking_mode,
        "max_size": self.config.max_size,
        "min_size": self.config.min_size,
        "buffer_size": self.config.buffer_size,
        "breakpoint_threshold_type": self.config.breakpoint_threshold_type,
        "breakpoint_threshold_amount": self.config.breakpoint_threshold_amount,
    }

    # Check cache first
    cached_result = self.cache.get(doc, config=config_dict)
    if cached_result is not None:
        logger.debug("Cache hit for document chunking")
        return cached_result

    # Perform chunking
    if self.chunking_mode == "naive":
        result = self._naive_chunk(doc)
    else:
        # Semantic chunking (requires sentence-transformers)
        if not SEMANTIC_CHUNKING_AVAILABLE:
            logger.warning(
                "Semantic chunking requested but not available. Falling back to naive chunking."
            )
            result = self._naive_chunk(doc)
        else:
            self._init_model()
            documents = [doc]

            if self._model is None or self._model is False:
                logger.warning(
                    "Model not initialized. Falling back to naive chunking."
                )
                result = self._naive_chunk(doc)
            elif SemanticChunker is None:  # type: ignore
                logger.warning(
                    "SemanticChunker not available. Falling back to naive chunking."
                )
                result = self._naive_chunk(doc)
            else:
                text_splitter = SemanticChunker(
                    buffer_size=self.config.buffer_size,
                    breakpoint_threshold_type=self.config.breakpoint_threshold_type,
                    breakpoint_threshold_amount=self.config.breakpoint_threshold_amount,
                    embeddings=self._model,
                    min_chunk_size=self.config.min_size,
                    sentence_split_regex=r"(?:(?:\n{2,}(?=#+))|(?:\n{2,}(?=- ))"
                    r"|(?<=[a-z][.?!])\s+(?=\b[A-Z]\w{8,}\b)|(?<!#)(?=#+))",
                )

                def recursive_chunking(docs_list: list[str], stop_flag=False):
                    lens = [len(d) for d in docs_list]
                    logger.info(f"chunk lengths: {lens}")
                    if (
                        all(len(d) < self.config.max_size for d in docs_list)
                        or stop_flag
                    ):
                        return docs_list
                    else:
                        new_docs = []
                        for d in docs_list:
                            if len(d) > self.config.max_size:
                                cdocs_ = text_splitter.create_documents([d])
                                cdocs = [d.page_content for d in cdocs_]
                                if len(cdocs[-1]) < self.config.min_size:
                                    cdocs = cdocs[:-2] + [cdocs[-2] + cdocs[-1]]
                                new_docs.extend(cdocs)
                            else:
                                new_docs.append(d)
                        stop_flag = len(docs_list) == len(new_docs)
                        return recursive_chunking(new_docs, stop_flag=stop_flag)

                result = recursive_chunking(documents)

    # Cache the result
    self.cache.set(doc, result, config=config_dict)
    logger.debug("Cached document chunking result")

    return result

__init__(chunk_config=None, cache=None, **kwargs)

Initialize the ChunkerTool.

Parameters:

Name Type Description Default
chunk_config ChunkConfig | None

Chunking configuration. If None, uses default ChunkConfig.

 None
cache Cacher | None

Optional shared Cacher instance. If None, creates a new one.

 None
**kwargs

Additional keyword arguments passed to the parent class.

 {}
Source code in ontocast/tool/chunk/chunker.py 
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
def __init__(
    self,
    chunk_config: ChunkConfig | None = None,
    cache: Cacher | None = None,
    **kwargs,
):
    """Initialize the ChunkerTool.

    Args:
        chunk_config: Chunking configuration. If None, uses default ChunkConfig.
        cache: Optional shared Cacher instance. If None, creates a new one.
        **kwargs: Additional keyword arguments passed to the parent class.
    """
    super().__init__(**kwargs)
    self._model = None
    self._model_lock = threading.Lock()  # Lock for thread-safe model initialization

    # Initialize cache - use shared cacher or create new one
    if cache is not None:
        self.cache = ToolCacher(cache, "chunker")
    else:
        # Fallback for backward compatibility
        shared_cache = Cacher()
        self.cache = ToolCacher(shared_cache, "chunker")

    # Override config if provided
    if chunk_config is not None:
        self.config = chunk_config

    # Override chunking mode if semantic chunking is not available
    if not SEMANTIC_CHUNKING_AVAILABLE and self.chunking_mode == "semantic":
        self.chunking_mode = "naive"
        logger.warning(
            "Semantic chunking not available (sentence-transformers not installed). "
            "Falling back to naive chunking."
        )

ConverterTool

Bases: Tool

Tool for converting documents to structured data.

This class provides functionality for converting various document formats into structured data that can be processed by the OntoCast system. It includes caching to avoid re-converting the same documents.

Attributes:

Name Type Description
supported_extensions set[str]

Set of supported file extensions.
cache Any

Cacher instance for caching conversion results.
Source code in ontocast/tool/converter.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
class ConverterTool(Tool):
    """Tool for converting documents to structured data.

    This class provides functionality for converting various document formats
    into structured data that can be processed by the OntoCast system.
    It includes caching to avoid re-converting the same documents.

    Attributes:
        supported_extensions: Set of supported file extensions.
        cache: Cacher instance for caching conversion results.
    """

    supported_extensions: set[str] = Field(
        default={".pdf", ".ppt", ".pptx"},
        description="Set of supported file extensions",
    )
    cache: Any = Field(default=None, exclude=True)

    def __init__(
        self,
        cache: Cacher | None = None,
        **kwargs,
    ):
        """Initialize the converter tool.

        Args:
            cache: Optional shared Cacher instance. If None, creates a new one.
            **kwargs: Additional keyword arguments passed to the parent class.
        """
        super().__init__(**kwargs)
        self._converter = None
        self._converter_lock = threading.Lock()  # Lock for thread-safe converter access

        # Initialize cache - use shared cacher or create new one
        if cache is not None:
            self.cache = ToolCacher(cache, "converter")
        else:
            # Fallback for backward compatibility
            shared_cache = Cacher()
            self.cache = ToolCacher(shared_cache, "converter")

        try:
            from docling.document_converter import DocumentConverter  # type: ignore

            self._converter = DocumentConverter()
        except ImportError as e:
            logger.error(f"Could not import DocumentConverter: {e}")

    def __call__(self, file_input: Union[bytes, str, pathlib.Path]) -> dict[str, Any]:
        """Convert a document to structured data.

        Args:
            file_input: The input file as either bytes, string, or pathlib.Path.

        Returns:
            dict[str, Any]: The converted document data.
        """
        # For plain text input, no caching needed
        if isinstance(file_input, str):
            return {"text": file_input}

        # Prepare content for caching
        if isinstance(file_input, bytes):
            content_for_cache = file_input
        elif isinstance(file_input, pathlib.Path):
            content_for_cache = file_input.read_bytes()
        else:
            # Fallback for other types
            return {"text": str(file_input)}

        # Check cache first
        cached_result = self.cache.get(content_for_cache)
        if cached_result is not None:
            logger.debug("Cache hit for document conversion")
            return cached_result

        # Convert document (with thread-safe access to converter)
        with self._converter_lock:
            if isinstance(file_input, bytes):
                if self._converter is None:
                    raise ImportError("DocumentConverter not available")
                try:
                    from docling.datamodel.base_models import (  # type: ignore
                        DocumentStream,
                    )

                    ds = DocumentStream(name="doc", stream=BytesIO(file_input))
                except ImportError:
                    raise ImportError(
                        f"Could not import DocumentConverter: {file_input}"
                    )
                result = self._converter.convert(ds)
                doc = result.document.export_to_markdown()
                converted_result = {"text": doc}
            elif isinstance(file_input, pathlib.Path):
                if self._converter is None:
                    raise ImportError(
                        f"Could not import DocumentConverter: {file_input}"
                    )
                result = self._converter.convert(file_input)
                doc = result.document.export_to_markdown()
                converted_result = {"text": doc}
            else:
                # Fallback for other types
                converted_result = {"text": str(file_input)}

        # Cache the result
        self.cache.set(content_for_cache, converted_result)
        logger.debug("Cached document conversion result")

        return converted_result

__call__(file_input)

Convert a document to structured data.

Parameters:

Name Type Description Default
file_input Union[bytes, str, Path]

The input file as either bytes, string, or pathlib.Path.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: The converted document data.
Source code in ontocast/tool/converter.py 
 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
def __call__(self, file_input: Union[bytes, str, pathlib.Path]) -> dict[str, Any]:
    """Convert a document to structured data.

    Args:
        file_input: The input file as either bytes, string, or pathlib.Path.

    Returns:
        dict[str, Any]: The converted document data.
    """
    # For plain text input, no caching needed
    if isinstance(file_input, str):
        return {"text": file_input}

    # Prepare content for caching
    if isinstance(file_input, bytes):
        content_for_cache = file_input
    elif isinstance(file_input, pathlib.Path):
        content_for_cache = file_input.read_bytes()
    else:
        # Fallback for other types
        return {"text": str(file_input)}

    # Check cache first
    cached_result = self.cache.get(content_for_cache)
    if cached_result is not None:
        logger.debug("Cache hit for document conversion")
        return cached_result

    # Convert document (with thread-safe access to converter)
    with self._converter_lock:
        if isinstance(file_input, bytes):
            if self._converter is None:
                raise ImportError("DocumentConverter not available")
            try:
                from docling.datamodel.base_models import (  # type: ignore
                    DocumentStream,
                )

                ds = DocumentStream(name="doc", stream=BytesIO(file_input))
            except ImportError:
                raise ImportError(
                    f"Could not import DocumentConverter: {file_input}"
                )
            result = self._converter.convert(ds)
            doc = result.document.export_to_markdown()
            converted_result = {"text": doc}
        elif isinstance(file_input, pathlib.Path):
            if self._converter is None:
                raise ImportError(
                    f"Could not import DocumentConverter: {file_input}"
                )
            result = self._converter.convert(file_input)
            doc = result.document.export_to_markdown()
            converted_result = {"text": doc}
        else:
            # Fallback for other types
            converted_result = {"text": str(file_input)}

    # Cache the result
    self.cache.set(content_for_cache, converted_result)
    logger.debug("Cached document conversion result")

    return converted_result

__init__(cache=None, **kwargs)

Initialize the converter tool.

Parameters:

Name Type Description Default
cache Cacher | None

Optional shared Cacher instance. If None, creates a new one.

 None
**kwargs

Additional keyword arguments passed to the parent class.

 {}
Source code in ontocast/tool/converter.py 
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
def __init__(
    self,
    cache: Cacher | None = None,
    **kwargs,
):
    """Initialize the converter tool.

    Args:
        cache: Optional shared Cacher instance. If None, creates a new one.
        **kwargs: Additional keyword arguments passed to the parent class.
    """
    super().__init__(**kwargs)
    self._converter = None
    self._converter_lock = threading.Lock()  # Lock for thread-safe converter access

    # Initialize cache - use shared cacher or create new one
    if cache is not None:
        self.cache = ToolCacher(cache, "converter")
    else:
        # Fallback for backward compatibility
        shared_cache = Cacher()
        self.cache = ToolCacher(shared_cache, "converter")

    try:
        from docling.document_converter import DocumentConverter  # type: ignore

        self._converter = DocumentConverter()
    except ImportError as e:
        logger.error(f"Could not import DocumentConverter: {e}")

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. It reads and writes ontologies and facts as Turtle (.ttl) files in specified directories.

The manager supports: - Loading ontologies from a dedicated ontology directory - Storing ontologies with versioned filenames - Storing facts with customizable filenames based on specifications - Error handling for file operations

Attributes:

Name Type Description
working_directory Path | None

Path to the working directory for storing data.
ontology_path Path | None

Optional path to the ontology directory for loading ontologies.
Source code in ontocast/tool/triple_manager/filesystem_manager.py 
 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
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
144
145
146
147
148
149
150
151
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. It reads and writes ontologies
    and facts as Turtle (.ttl) files in specified directories.

    The manager supports:
    - Loading ontologies from a dedicated ontology directory
    - Storing ontologies with versioned filenames
    - Storing facts with customizable filenames based on specifications
    - Error handling for file operations

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

    working_directory: pathlib.Path | None
    ontology_path: pathlib.Path | None

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

        This method sets up the filesystem manager with the specified
        working and ontology directories.

        Args:
            **kwargs: Additional keyword arguments passed to the parent class.
                working_directory: Path to the working directory for storing data.
                ontology_path: Path to the ontology directory for loading ontologies.

        Example:
            >>> manager = FilesystemTripleStoreManager(
            ...     working_directory="/path/to/work",
            ...     ontology_path="/path/to/ontologies"
            ... )
        """
        super().__init__(**kwargs)

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

        This method scans the ontology directory for Turtle (.ttl) files
        and loads each one as an Ontology object. Files are processed
        in sorted order for consistent results.

        Returns:
            list[Ontology]: List of all ontologies found in the ontology directory.

        Example:
            >>> ontologies = manager.fetch_ontologies()
            >>> for onto in ontologies:
            ...     print(f"Loaded ontology: {onto.ontology_id}")
        """
        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)
                    logger.debug(f"Successfully loaded ontology from {fname}")
                except Exception as e:
                    logger.error(f"Failed to load ontology {fname}: {str(e)}")
        return ontologies

    def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
        """Store an RDF graph in the filesystem.

        This method stores the given RDF graph as a Turtle file in the
        working directory. The filename is generated based on the graph_uri
        parameter or defaults to "current.ttl".

        Args:
            graph: The RDF graph to store.
            fname:  str

        Example:
            >>> graph = RDFGraph()
            >>> manager.serialize_graph(graph)
            # Creates: working_directory/current.ttl

            >>> manager.serialize_graph(graph, fname="facts_abc.ttl")
        """
        if self.working_directory is None:
            return

        fname: str = kwargs.pop("fname")
        output_path = self.working_directory / fname
        graph.serialize(format="turtle", destination=output_path)
        logger.info(f"Graph saved to {output_path}")

    def serialize(self, o: Ontology | RDFGraph, graph_uri: str | None = None):
        if isinstance(o, Ontology):
            graph = o.graph
            fname = f"ontology_{o.ontology_id}_{o.version}.ttl"
        elif isinstance(o, RDFGraph):
            graph = o
            if graph_uri:
                s = graph_uri.split("/")[-2:]
                s = "_".join([x for x in s if x])
                fname = f"facts_{s}.ttl"
            else:
                fname = "facts_default.ttl"
        else:
            raise TypeError(f"unsupported obj of type {type(o)} received")

        self.serialize_graph(graph=graph, fname=fname)

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

        This method deletes all Turtle (.ttl) files from both the working
        directory and the ontology directory.

        Args:
            dataset: Optional dataset parameter (ignored for Filesystem, which doesn't
                support datasets). Included for interface compatibility.

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

        Raises:
            Exception: If the cleanup operation fails.
        """
        if dataset is not None:
            logger.warning(
                f"Dataset parameter '{dataset}' ignored for Filesystem (datasets not supported)"
            )
            logger.warning(
                "clean method not implemented for FilesystemTripleStoreManager"
            )

__init__(**kwargs)

Initialize the filesystem triple store manager.

This method sets up the filesystem manager with the specified working and ontology directories.

Parameters:

Name Type Description Default
**kwargs

Additional keyword arguments passed to the parent class. working_directory: Path to the working directory for storing data. ontology_path: Path to the ontology directory for loading ontologies.

 {}
Example

manager = FilesystemTripleStoreManager( ... working_directory="/path/to/work", ... ontology_path="/path/to/ontologies" ... )

Source code in ontocast/tool/triple_manager/filesystem_manager.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(self, **kwargs):
    """Initialize the filesystem triple store manager.

    This method sets up the filesystem manager with the specified
    working and ontology directories.

    Args:
        **kwargs: Additional keyword arguments passed to the parent class.
            working_directory: Path to the working directory for storing data.
            ontology_path: Path to the ontology directory for loading ontologies.

    Example:
        >>> manager = FilesystemTripleStoreManager(
        ...     working_directory="/path/to/work",
        ...     ontology_path="/path/to/ontologies"
        ... )
    """
    super().__init__(**kwargs)

clean(dataset=None) async

Clean/flush all data from the filesystem triple store.

This method deletes all Turtle (.ttl) files from both the working directory and the ontology directory.

Parameters:

Name Type Description Default
dataset str | None

Optional dataset parameter (ignored for Filesystem, which doesn't support datasets). Included for interface compatibility.

 None

Raises:

Type Description
Exception

If the cleanup operation fails.
Source code in ontocast/tool/triple_manager/filesystem_manager.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
async def clean(self, dataset: str | None = None) -> None:
    """Clean/flush all data from the filesystem triple store.

    This method deletes all Turtle (.ttl) files from both the working
    directory and the ontology directory.

    Args:
        dataset: Optional dataset parameter (ignored for Filesystem, which doesn't
            support datasets). Included for interface compatibility.

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

    Raises:
        Exception: If the cleanup operation fails.
    """
    if dataset is not None:
        logger.warning(
            f"Dataset parameter '{dataset}' ignored for Filesystem (datasets not supported)"
        )
        logger.warning(
            "clean method not implemented for FilesystemTripleStoreManager"
        )

fetch_ontologies()

Fetch all available ontologies from the filesystem.

This method scans the ontology directory for Turtle (.ttl) files and loads each one as an Ontology object. Files are processed in sorted order for consistent results.

Returns:

Type Description
list[Ontology]

list[Ontology]: List of all ontologies found in the ontology directory.
Example

ontologies = manager.fetch_ontologies() for onto in ontologies: ... print(f"Loaded ontology: {onto.ontology_id}")

Source code in ontocast/tool/triple_manager/filesystem_manager.py 
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
def fetch_ontologies(self) -> list[Ontology]:
    """Fetch all available ontologies from the filesystem.

    This method scans the ontology directory for Turtle (.ttl) files
    and loads each one as an Ontology object. Files are processed
    in sorted order for consistent results.

    Returns:
        list[Ontology]: List of all ontologies found in the ontology directory.

    Example:
        >>> ontologies = manager.fetch_ontologies()
        >>> for onto in ontologies:
        ...     print(f"Loaded ontology: {onto.ontology_id}")
    """
    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)
                logger.debug(f"Successfully loaded ontology from {fname}")
            except Exception as e:
                logger.error(f"Failed to load ontology {fname}: {str(e)}")
    return ontologies

serialize_graph(graph, **kwargs)

Store an RDF graph in the filesystem.

This method stores the given RDF graph as a Turtle file in the working directory. The filename is generated based on the graph_uri parameter or defaults to "current.ttl".

Parameters:

Name Type Description Default
graph Graph

The RDF graph to store.

 required
fname

str

 required
Example

graph = RDFGraph() manager.serialize_graph(graph)

Creates: working_directory/current.ttl

manager.serialize_graph(graph, fname="facts_abc.ttl")

Source code in ontocast/tool/triple_manager/filesystem_manager.py 
 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
def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
    """Store an RDF graph in the filesystem.

    This method stores the given RDF graph as a Turtle file in the
    working directory. The filename is generated based on the graph_uri
    parameter or defaults to "current.ttl".

    Args:
        graph: The RDF graph to store.
        fname:  str

    Example:
        >>> graph = RDFGraph()
        >>> manager.serialize_graph(graph)
        # Creates: working_directory/current.ttl

        >>> manager.serialize_graph(graph, fname="facts_abc.ttl")
    """
    if self.working_directory is None:
        return

    fname: str = kwargs.pop("fname")
    output_path = self.working_directory / fname
    graph.serialize(format="turtle", destination=output_path)
    logger.info(f"Graph saved to {output_path}")

FusekiTripleStoreManager

Bases: TripleStoreManagerWithAuth

Fuseki-based triple store manager.

This class provides a concrete implementation of triple store management using Apache Fuseki. It stores ontologies as named graphs using their URIs as graph names, and supports dataset creation and cleanup.

The manager uses Fuseki's REST API for all operations, including: - Dataset creation and management - Named graph operations for ontologies - SPARQL queries for ontology discovery - Graph-level data operations

Attributes:

Name Type Description
dataset str | None

The Fuseki dataset name to use for storage.
clean None

Whether to clean the dataset on initialization.
Source code in ontocast/tool/triple_manager/fuseki.py 
 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
144
145
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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
class FusekiTripleStoreManager(TripleStoreManagerWithAuth):
    """Fuseki-based triple store manager.

    This class provides a concrete implementation of triple store management
    using Apache Fuseki. It stores ontologies as named graphs using their
    URIs as graph names, and supports dataset creation and cleanup.

    The manager uses Fuseki's REST API for all operations, including:
    - Dataset creation and management
    - Named graph operations for ontologies
    - SPARQL queries for ontology discovery
    - Graph-level data operations

    Attributes:
        dataset: The Fuseki dataset name to use for storage.
        clean: Whether to clean the dataset on initialization.
    """

    dataset: str | None = Field(default=None, description="Fuseki dataset name")
    ontologies_dataset: str = Field(
        default=DEFAULT_ONTOLOGIES_DATASET,
        description="Fuseki dataset name for ontologies",
    )

    def __init__(
        self,
        uri=None,
        auth=None,
        dataset=None,
        ontologies_dataset=None,
        **kwargs,
    ):
        """Initialize the Fuseki triple store manager.

        This method sets up the connection to Fuseki and creates the dataset
        if it doesn't exist. The dataset is NOT cleaned on initialization.

        Args:
            uri: Fuseki server URI (e.g., "http://localhost:3030").
            auth: Authentication tuple (username, password) or string in "user/password" format.
            dataset: Dataset name to use for storage.
            ontologies_dataset: Dataset name for ontologies (defaults to separate dataset).
            **kwargs: Additional keyword arguments passed to the parent class.

        Raises:
            ValueError: If dataset is not specified in URI or as argument.

        Example:
            >>> manager = FusekiTripleStoreManager(
            ...     uri="http://localhost:3030",
            ...     dataset="test"
            ... )
            >>> # To clean the dataset, use the clean() method explicitly:
            >>> await manager.clean()
        """
        super().__init__(
            uri=uri, auth=auth, env_uri="FUSEKI_URI", env_auth="FUSEKI_AUTH", **kwargs
        )
        if dataset is None:
            self.dataset = DEFAULT_DATASET
        else:
            self.dataset = dataset
        self.ontologies_dataset = ontologies_dataset or DEFAULT_ONTOLOGIES_DATASET

        # Initialize httpx client for async operations
        self._client: httpx.AsyncClient | None = None

        # Initialize datasets synchronously (for backward compatibility)
        # In async contexts, use async_init() instead
        asyncio.run(self._async_init_with_cleanup())

    async def _async_init_with_cleanup(self):
        """Wrapper for async_init that ensures proper cleanup when using asyncio.run().

        This method creates a temporary client and ensures it's properly closed
        before returning, preventing "Event loop is closed" errors.
        """
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                await self._async_init()
            finally:
                # Restore original client
                self._client = original_client

    async def _async_init(self):
        """Async initialization of datasets."""
        await self.init_dataset(self.dataset)
        if self.ontologies_dataset != self.dataset:
            await self.init_dataset(self.ontologies_dataset)

    def _prepare_auth(self) -> httpx.BasicAuth | None:
        """Prepare httpx BasicAuth from self.auth.

        Returns:
            httpx.BasicAuth instance or None if no auth is configured.
        """
        if self.auth:
            if isinstance(self.auth, tuple):
                return httpx.BasicAuth(*self.auth)
            elif isinstance(self.auth, str) and "/" in self.auth:
                parts = self.auth.split("/", 1)
                if len(parts) == 2:
                    username, password = parts[0], parts[1]
                    return httpx.BasicAuth(username, password)
        return None

    async def _get_client(self) -> httpx.AsyncClient:
        """Get or create the httpx async client."""
        if self._client is None:
            auth = self._prepare_auth()
            self._client = httpx.AsyncClient(auth=auth, timeout=30.0)
        return self._client

    async def close(self):
        """Close the httpx client."""
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    async def update_dataset(self, new_dataset: str) -> None:
        """Update the dataset name for this manager.

        This method allows changing the dataset without recreating the entire
        manager, which is useful for API requests that specify different datasets.

        Args:
            new_dataset: The new dataset name to use.
        """
        if not new_dataset:
            raise ValueError("Dataset name cannot be empty")

        self.dataset = new_dataset
        await self.init_dataset(self.dataset)
        logger.info(f"Updated Fuseki dataset to: {self.dataset}")

    async def clean(self, dataset: str | None = None) -> None:
        """Clean/flush data from Fuseki dataset(s).

        This method removes all named graphs and clears the default graph
        from the specified dataset, or all datasets if no dataset is specified.

        Args:
            dataset: Optional dataset name to clean. If None, cleans both the main
                dataset and the ontologies dataset. If specified, cleans only that dataset.

        Warning: This operation is irreversible and will delete all data
        from the specified dataset(s).

        The method handles errors gracefully and logs the results of
        each cleanup operation.

        Example:
            >>> # Clean all datasets
            >>> await manager.clean()
            >>> # Clean specific dataset
            >>> await manager.clean(dataset="my_dataset")
        """
        if dataset is None:
            # Clean all datasets (main and ontologies)
            # self.dataset is guaranteed to be a string (set to DEFAULT_DATASET if None in __init__)
            assert self.dataset is not None, "Dataset should never be None"
            await self._clean_dataset_by_name(self.dataset)
            logger.info(f"Fuseki dataset '{self.dataset}' cleaned (all data deleted)")

            # Also clean the ontologies dataset if it's different
            if self.ontologies_dataset != self.dataset:
                await self._clean_dataset_by_name(self.ontologies_dataset)
                logger.info(
                    f"Fuseki ontologies dataset '{self.ontologies_dataset}' cleaned (all data deleted)"
                )
        else:
            # Clean only the specified dataset
            await self._clean_dataset_by_name(dataset)
            logger.info(f"Fuseki dataset '{dataset}' cleaned (all data deleted)")

    async def _clean_dataset_by_name(self, dataset_name: str) -> None:
        """Clean a specific dataset by name.

        This is a helper method that performs the actual cleaning of a single dataset.
        It deletes all named graphs and clears the default graph.

        Uses a temporary client to avoid event loop cleanup issues when called
        from different async contexts.

        Args:
            dataset_name: Name of the dataset to clean.

        Raises:
            Exception: If the cleanup operation fails.
        """
        # Use a temporary client to avoid event loop cleanup issues
        async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
            try:
                dataset_url = f"{self.uri}/{dataset_name}"
                sparql_update_url = f"{dataset_url}/update"
                sparql_url = f"{dataset_url}/sparql"

                # Delete all named graphs
                query = """
                SELECT DISTINCT ?g WHERE {
                  GRAPH ?g { ?s ?p ?o }
                }
                """
                response = await client.post(
                    sparql_url,
                    data={"query": query, "format": "application/sparql-results+json"},
                )

                if response.status_code == 200:
                    results = response.json()
                    tasks = []
                    for binding in results.get("results", {}).get("bindings", []):
                        graph_uri = binding["g"]["value"]
                        # Delete the named graph using SPARQL UPDATE
                        drop_query = f"DROP GRAPH <{graph_uri}>"
                        tasks.append(
                            client.post(
                                sparql_update_url,
                                data={"update": drop_query},
                            )
                        )

                    # Execute all deletions in parallel
                    delete_responses = await asyncio.gather(
                        *tasks, return_exceptions=True
                    )
                    for i, delete_response in enumerate(delete_responses):
                        graph_uri = results["results"]["bindings"][i]["g"]["value"]
                        if isinstance(delete_response, Exception):
                            logger.warning(
                                f"Failed to delete graph {graph_uri}: {delete_response}"
                            )
                        elif isinstance(delete_response, httpx.Response):
                            if delete_response.status_code in (200, 204):
                                logger.debug(f"Deleted named graph: {graph_uri}")
                            else:
                                logger.warning(
                                    f"Failed to delete graph {graph_uri}: {delete_response.status_code}"
                                )

                # Clear the default graph using SPARQL UPDATE
                clear_query = "CLEAR DEFAULT"
                clear_response = await client.post(
                    sparql_update_url,
                    data={"update": clear_query},
                )
                if clear_response.status_code in (200, 204):
                    logger.debug(f"Cleared default graph in dataset '{dataset_name}'")
                else:
                    logger.warning(
                        f"Failed to clear default graph in dataset '{dataset_name}': {clear_response.status_code}"
                    )
            except Exception as e:
                logger.error(f"Failed to clean dataset '{dataset_name}': {e}")
                raise

    async def init_dataset(self, dataset_name):
        """Initialize a Fuseki dataset.

        This method creates a new dataset in Fuseki if it doesn't already exist.
        It uses Fuseki's admin API to create the dataset with TDB2 storage.

        Uses a temporary client to avoid event loop cleanup issues when called
        from different async contexts.

        Args:
            dataset_name: Name of the dataset to create.

        Note:
            This method will not fail if the dataset already exists.
        """
        # Use a temporary client to avoid event loop cleanup issues
        async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
            fuseki_admin_url = f"{self.uri}/$/datasets"

            payload = {"dbName": dataset_name, "dbType": "tdb2"}

            headers = {"Content-Type": "application/x-www-form-urlencoded"}

            response = await client.post(
                fuseki_admin_url, data=payload, headers=headers
            )

            if response.status_code == 200 or response.status_code == 201:
                logger.info(f"Fuseki dataset '{dataset_name}' created successfully.")
            elif response.status_code == 409:
                logger.info(
                    f"Fuseki status code: {response.status_code}; {response.text.strip()}"
                )
            else:
                logger.error(
                    f"Failed to create dataset {dataset_name}. Status code: {response.status_code}"
                )
                logger.error(f"Response: {response.text.strip()}")

    def _get_dataset_url(self):
        """Get the full URL for the dataset.

        Returns:
            str: The complete URL for the dataset endpoint.
        """
        return f"{self.uri}/{self.dataset}"

    def _get_ontologies_dataset_url(self):
        """Get the full URL for the ontologies dataset.

        Returns:
            str: The complete URL for the ontologies dataset endpoint.
        """
        return f"{self.uri}/{self.ontologies_dataset}"

    def fetch_ontologies(self) -> list[Ontology]:
        """Synchronous wrapper for fetch_ontologies.

        For async usage, use afetch_ontologies() instead.
        """
        # Use a temporary client for this operation to avoid event loop cleanup issues
        return asyncio.run(self._fetch_ontologies_with_cleanup())

    async def afetch_ontologies(self) -> list[Ontology]:
        """Async version of fetch_ontologies.

        This is the preferred method when running in an async context.
        """
        return await self._fetch_ontologies_async()

    async def _fetch_ontologies_with_cleanup(self) -> list[Ontology]:
        """Wrapper that ensures proper cleanup when using asyncio.run().

        This method creates a temporary client and ensures it's properly closed
        before returning, preventing "Event loop is closed" errors.
        """
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                return await self._fetch_ontologies_async()
            finally:
                # Restore original client
                self._client = original_client

    async def _fetch_ontologies_async(self) -> list[Ontology]:
        """Fetch all ontologies from their corresponding named graphs.

        This method discovers all ontologies in the Fuseki ontologies dataset and
        fetches each one from its corresponding named graph. For versioned ontologies,
        it returns only the latest version for each unique ontology IRI.

        1. Discovery: List all named graphs (which may be versioned URIs)
        2. Fetching: Retrieve each ontology from its named graph (in parallel)
        3. Deduplication: For versioned ontologies, keep only the latest version

        Returns:
            list[Ontology]: List of the latest version of each ontology found.

        Example:
            >>> ontologies = await manager.fetch_ontologies()
            >>> for onto in ontologies:
            ...     print(f"Found ontology: {onto.iri} v{onto.version}")
        """
        client = await self._get_client()
        sparql_url = f"{self._get_ontologies_dataset_url()}/sparql"

        # Step 1: List all named graphs
        list_query = """
        SELECT DISTINCT ?g WHERE {
          GRAPH ?g { ?s ?p ?o }
        }
        """
        response = await client.post(
            sparql_url,
            data={"query": list_query, "format": "application/sparql-results+json"},
        )
        if response.status_code != 200:
            logger.error(f"Failed to list graphs from Fuseki: {response.text}")
            return []

        results = response.json()
        graph_uris = []
        for binding in results.get("results", {}).get("bindings", []):
            graph_uri = binding["g"]["value"]
            graph_uris.append(graph_uri)

        logger.debug(f"Found {len(graph_uris)} named graphs: {graph_uris}")

        # Step 2: Fetch each ontology from its corresponding named graph (in parallel)
        async def fetch_single_ontology(graph_uri: str) -> Ontology | None:
            """Fetch a single ontology from a graph URI."""
            try:
                graph = RDFGraph()
                # URL encode the graph URI to handle special characters like #
                encoded_graph_uri = quote(str(graph_uri), safe="/:")
                export_url = f"{self._get_ontologies_dataset_url()}/get?graph={encoded_graph_uri}"
                export_resp = await client.get(
                    export_url, headers={"Accept": "text/turtle"}
                )

                if export_resp.status_code == 200:
                    graph.parse(data=export_resp.text, format="turtle")

                    # Re-serialize deterministically to ensure consistent cache keys
                    # This sorts both namespaces and triples alphabetically
                    deterministic_turtle = deterministic_turtle_serialization(graph)

                    # Re-parse from deterministic serialization to ensure we have RDFGraph
                    deterministic_graph = RDFGraph()
                    deterministic_graph.parse(
                        data=deterministic_turtle, format="turtle"
                    )

                    # Copy namespace bindings from original graph
                    for prefix, namespace in graph.namespaces():
                        if prefix:
                            deterministic_graph.bind(prefix, namespace)

                    graph = deterministic_graph

                    # Find the ontology IRI in the graph
                    for onto_subj, _, obj in graph.triples(
                        (None, RDF.type, OWL.Ontology)
                    ):
                        onto_iri = str(onto_subj)
                        # Extract base IRI if graph_uri is versioned
                        # Handle both hash fragments (#19193944...) and semantic versions (#v1.2.3)
                        if "#" in graph_uri:
                            base_iri = graph_uri.split("#")[0]
                            # Use base IRI from graph_uri (named graph identifier)
                            # The graph content should have simplified IRI, but use graph_uri as source of truth
                            onto_iri = base_iri

                        ontology = Ontology(
                            graph=graph,
                            iri=onto_iri,
                        )
                        # Load properties from graph (will strip any hash fragments if present)
                        ontology.sync_properties_from_graph()
                        logger.debug(
                            f"Successfully loaded ontology: {onto_iri} version: {ontology.version}"
                        )
                        return ontology
                else:
                    logger.warning(
                        f"Failed to fetch graph {graph_uri}: {export_resp.status_code}"
                    )
            except Exception as e:
                logger.warning(f"Error fetching ontology from {graph_uri}: {e}")
            return None

        # Fetch all ontologies in parallel
        all_ontologies_results = await asyncio.gather(
            *[fetch_single_ontology(uri) for uri in graph_uris], return_exceptions=True
        )

        # Filter out None and exceptions
        all_ontologies = []
        for result in all_ontologies_results:
            if isinstance(result, Exception):
                logger.warning(f"Exception fetching ontology: {result}")
            elif result is not None:
                all_ontologies.append(result)

        # Step 3: Deduplicate and keep latest terminal versions
        ontology_dict = defaultdict(list)

        for onto in all_ontologies:
            ontology_dict[onto.iri].append(onto)

        # Build set of all parent hashes to identify terminal ontologies
        # A terminal ontology is one that is not a parent for any other ontology
        all_parent_hashes = set()

        for onto in all_ontologies:
            if onto.hash:
                # Collect all parent hashes
                for parent_hash in onto.parent_hashes:
                    all_parent_hashes.add(parent_hash)

        # For each unique IRI, select the latest terminal ontology
        ontologies = []

        for iri, versions in ontology_dict.items():
            if len(versions) == 1:
                ontologies.append(versions[0])
            else:
                # Multiple versions - find terminal ontologies (not parents)
                terminal_versions = [
                    v for v in versions if v.hash and v.hash not in all_parent_hashes
                ]

                if not terminal_versions:
                    # No terminal ontologies found - all are parents
                    # Fall back to non-terminal versions
                    logger.warning(
                        f"No terminal ontologies found for {iri}, "
                        f"using all versions for selection"
                    )
                    terminal_versions = versions

                # Select latest by created_at among terminal ontologies
                try:
                    versions_with_created = [
                        v for v in terminal_versions if v.created_at is not None
                    ]

                    if versions_with_created:
                        # Sort by created_at (most recent first)
                        versions_with_created.sort(
                            key=lambda x: x.created_at, reverse=True
                        )
                        selected = versions_with_created[0]
                        hash_str = (
                            f"{selected.hash[:16]}..." if selected.hash else "no hash"
                        )
                        logger.debug(
                            f"Selected terminal ontology for {iri} "
                            f"by created_at: {selected.created_at} "
                            f"(hash: {hash_str})"
                        )
                        ontologies.append(selected)
                    else:
                        # No created_at available - fall back to version-based sorting
                        versions_with_ver = [v for v in terminal_versions if v.version]
                        if versions_with_ver:
                            versions_with_ver.sort(
                                key=lambda x: str(x.version), reverse=False
                            )
                            selected = versions_with_ver[-1]
                            logger.debug(
                                f"Selected terminal ontology for {iri} "
                                f"by version: {selected.version} "
                                f"(no created_at available)"
                            )
                            ontologies.append(selected)
                        else:
                            # No version info either - use first terminal ontology
                            selected = terminal_versions[0]
                            logger.debug(
                                f"Selected first terminal ontology for {iri} "
                                f"(no created_at or version available)"
                            )
                            ontologies.append(selected)
                except Exception as e:
                    logger.warning(
                        f"Could not select terminal ontology for {iri}: {e}, "
                        f"using first version"
                    )
                    ontologies.append(terminal_versions[0])

        logger.info(
            f"Successfully loaded {len(ontologies)} unique ontologies from Fuseki "
        )
        return ontologies

    def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
        """Synchronous wrapper for serialize_graph.

        For async usage, use aserialize_graph() instead.
        """
        return asyncio.run(self._serialize_graph_with_cleanup(graph, **kwargs))

    async def aserialize_graph(self, graph: Graph, **kwargs) -> bool | None:
        """Async version of serialize_graph.

        This is the preferred method when running in an async context.
        """
        return await self._serialize_graph_async(graph, **kwargs)

    async def _serialize_graph_with_cleanup(
        self, graph: Graph, **kwargs
    ) -> bool | None:
        """Wrapper that ensures proper cleanup when using asyncio.run().

        This method creates a temporary client and ensures it's properly closed
        before returning, preventing "Event loop is closed" errors.
        """
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                return await self._serialize_graph_async(graph, **kwargs)
            finally:
                # Restore original client
                self._client = original_client

    async def _serialize_graph_async(self, graph: Graph, **kwargs) -> bool | None:
        """Store an RDF graph as a named graph in a specific Fuseki dataset.

        This is a private helper method that handles the common logic for storing
        graphs in Fuseki datasets.

        Args:
            graph: The RDF graph to store.
            **kwargs: Additional parameters including graph_uri, dataset_url, default_graph_uri, log_prefix.

        Returns:
            bool: True if the graph was successfully stored, False otherwise.
        """
        client = await self._get_client()
        graph_uri = kwargs.get("graph_uri")
        dataset_url = kwargs.get("dataset_url")
        default_graph_uri = kwargs.get("default_graph_uri")
        log_prefix = kwargs.get("log_prefix")

        turtle_data = graph.serialize(format="turtle")
        if graph_uri is None:
            graph_uri = default_graph_uri

        # URL encode the graph URI to handle special characters like #
        encoded_graph_uri = quote(str(graph_uri), safe="/:")
        url = f"{dataset_url}/data?graph={encoded_graph_uri}"
        headers = {"Content-Type": "text/turtle;charset=utf-8"}
        response = await client.put(url, headers=headers, content=turtle_data)
        if response.status_code in (200, 201, 204):
            logger.info(
                f"{log_prefix} graph {graph_uri} uploaded to Fuseki as named graph."
            )
            return True
        else:
            logger.error(
                f"Failed to upload {log_prefix.lower() if log_prefix else 'unknown'} graph {graph_uri}. Status code: {response.status_code}"
            )
            logger.error(f"Response: {response.text}")
            return False

    def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
        """Synchronous wrapper for serialize.

        For async usage, use aserialize() instead.
        """
        return asyncio.run(self._serialize_with_cleanup(o, **kwargs))

    async def aserialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
        """Async version of serialize.

        This is the preferred method when running in an async context.
        """
        return await self._serialize_async(o, **kwargs)

    async def _serialize_with_cleanup(
        self, o: Ontology | RDFGraph, **kwargs
    ) -> bool | None:
        """Wrapper that ensures proper cleanup when using asyncio.run().

        This method creates a temporary client and ensures it's properly closed
        before returning, preventing "Event loop is closed" errors.
        """
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                return await self._serialize_async(o, **kwargs)
            finally:
                # Restore original client
                self._client = original_client

    async def _serialize_async(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
        """Store an RDF graph as a named graph in Fuseki.

        This method stores the given RDF graph as a named graph in Fuseki.
        The graph name is taken from the graph_uri parameter or defaults to
        "urn:data:default".

        Args:
            o: RDF graph or Ontology object.
            **kwargs: Additional parameters including graph_uri.

        Returns:
            bool: True if the graph was successfully stored, False otherwise.

        Example:
            >>> graph = RDFGraph()
            >>> success = await manager.serialize(graph)

            >>> success = await manager.serialize(graph, graph_uri="http://example.org/chunk1")
        """
        graph_uri = kwargs.get("graph_uri")

        if isinstance(o, Ontology):
            graph = o.graph
            # Use versioned IRI for storage to enable multiple versions to coexist
            graph_uri = o.versioned_iri
            default_graph_uri = "urn:ontology:default"
            log_prefix = "Ontology"
            # Use ontologies dataset for ontology storage
            dataset_url = self._get_ontologies_dataset_url()
        elif isinstance(o, RDFGraph):
            graph = o
            default_graph_uri = "urn:data:default"
            log_prefix = "Graph"
            # Use regular dataset for facts storage
            dataset_url = self._get_dataset_url()
        else:
            raise TypeError(f"unsupported obj of type {type(o)} received")

        return await self._serialize_graph_async(
            graph=graph,
            graph_uri=graph_uri,
            dataset_url=dataset_url,
            default_graph_uri=default_graph_uri,
            log_prefix=log_prefix,
        )

__init__(uri=None, auth=None, dataset=None, ontologies_dataset=None, **kwargs)

Initialize the Fuseki triple store manager.

This method sets up the connection to Fuseki and creates the dataset if it doesn't exist. The dataset is NOT cleaned on initialization.

Parameters:

Name Type Description Default
uri

Fuseki server URI (e.g., "http://localhost:3030").

 None
auth

Authentication tuple (username, password) or string in "user/password" format.

 None
dataset

Dataset name to use for storage.

 None
ontologies_dataset

Dataset name for ontologies (defaults to separate dataset).

 None
**kwargs

Additional keyword arguments passed to the parent class.

 {}

Raises:

Type Description
ValueError

If dataset is not specified in URI or as argument.
Example

manager = FusekiTripleStoreManager( ... uri="http://localhost:3030", ... dataset="test" ... )

To clean the dataset, use the clean() method explicitly:

await manager.clean()

Source code in ontocast/tool/triple_manager/fuseki.py 
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
160
161
162
163
164
165
166
def __init__(
    self,
    uri=None,
    auth=None,
    dataset=None,
    ontologies_dataset=None,
    **kwargs,
):
    """Initialize the Fuseki triple store manager.

    This method sets up the connection to Fuseki and creates the dataset
    if it doesn't exist. The dataset is NOT cleaned on initialization.

    Args:
        uri: Fuseki server URI (e.g., "http://localhost:3030").
        auth: Authentication tuple (username, password) or string in "user/password" format.
        dataset: Dataset name to use for storage.
        ontologies_dataset: Dataset name for ontologies (defaults to separate dataset).
        **kwargs: Additional keyword arguments passed to the parent class.

    Raises:
        ValueError: If dataset is not specified in URI or as argument.

    Example:
        >>> manager = FusekiTripleStoreManager(
        ...     uri="http://localhost:3030",
        ...     dataset="test"
        ... )
        >>> # To clean the dataset, use the clean() method explicitly:
        >>> await manager.clean()
    """
    super().__init__(
        uri=uri, auth=auth, env_uri="FUSEKI_URI", env_auth="FUSEKI_AUTH", **kwargs
    )
    if dataset is None:
        self.dataset = DEFAULT_DATASET
    else:
        self.dataset = dataset
    self.ontologies_dataset = ontologies_dataset or DEFAULT_ONTOLOGIES_DATASET

    # Initialize httpx client for async operations
    self._client: httpx.AsyncClient | None = None

    # Initialize datasets synchronously (for backward compatibility)
    # In async contexts, use async_init() instead
    asyncio.run(self._async_init_with_cleanup())

afetch_ontologies() async

Async version of fetch_ontologies.

This is the preferred method when running in an async context.

Source code in ontocast/tool/triple_manager/fuseki.py 
421
422
423
424
425
426
async def afetch_ontologies(self) -> list[Ontology]:
    """Async version of fetch_ontologies.

    This is the preferred method when running in an async context.
    """
    return await self._fetch_ontologies_async()

aserialize(o, **kwargs) async

Async version of serialize.

This is the preferred method when running in an async context.

Source code in ontocast/tool/triple_manager/fuseki.py 
739
740
741
742
743
744
async def aserialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
    """Async version of serialize.

    This is the preferred method when running in an async context.
    """
    return await self._serialize_async(o, **kwargs)

aserialize_graph(graph, **kwargs) async

Async version of serialize_graph.

This is the preferred method when running in an async context.

Source code in ontocast/tool/triple_manager/fuseki.py 
665
666
667
668
669
670
async def aserialize_graph(self, graph: Graph, **kwargs) -> bool | None:
    """Async version of serialize_graph.

    This is the preferred method when running in an async context.
    """
    return await self._serialize_graph_async(graph, **kwargs)

clean(dataset=None) async

Clean/flush data from Fuseki dataset(s).

This method removes all named graphs and clears the default graph from the specified dataset, or all datasets if no dataset is specified.

Parameters:

Name Type Description Default
dataset str | None

Optional dataset name to clean. If None, cleans both the main dataset and the ontologies dataset. If specified, cleans only that dataset.

 None

from the specified dataset(s).

The method handles errors gracefully and logs the results of each cleanup operation.

Example

Clean all datasets

await manager.clean()

Clean specific dataset

await manager.clean(dataset="my_dataset")

Source code in ontocast/tool/triple_manager/fuseki.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
async def clean(self, dataset: str | None = None) -> None:
    """Clean/flush data from Fuseki dataset(s).

    This method removes all named graphs and clears the default graph
    from the specified dataset, or all datasets if no dataset is specified.

    Args:
        dataset: Optional dataset name to clean. If None, cleans both the main
            dataset and the ontologies dataset. If specified, cleans only that dataset.

    Warning: This operation is irreversible and will delete all data
    from the specified dataset(s).

    The method handles errors gracefully and logs the results of
    each cleanup operation.

    Example:
        >>> # Clean all datasets
        >>> await manager.clean()
        >>> # Clean specific dataset
        >>> await manager.clean(dataset="my_dataset")
    """
    if dataset is None:
        # Clean all datasets (main and ontologies)
        # self.dataset is guaranteed to be a string (set to DEFAULT_DATASET if None in __init__)
        assert self.dataset is not None, "Dataset should never be None"
        await self._clean_dataset_by_name(self.dataset)
        logger.info(f"Fuseki dataset '{self.dataset}' cleaned (all data deleted)")

        # Also clean the ontologies dataset if it's different
        if self.ontologies_dataset != self.dataset:
            await self._clean_dataset_by_name(self.ontologies_dataset)
            logger.info(
                f"Fuseki ontologies dataset '{self.ontologies_dataset}' cleaned (all data deleted)"
            )
    else:
        # Clean only the specified dataset
        await self._clean_dataset_by_name(dataset)
        logger.info(f"Fuseki dataset '{dataset}' cleaned (all data deleted)")

close() async

Close the httpx client.

Source code in ontocast/tool/triple_manager/fuseki.py 
215
216
217
218
219
async def close(self):
    """Close the httpx client."""
    if self._client is not None:
        await self._client.aclose()
        self._client = None

fetch_ontologies()

Synchronous wrapper for fetch_ontologies.

For async usage, use afetch_ontologies() instead.

Source code in ontocast/tool/triple_manager/fuseki.py 
413
414
415
416
417
418
419
def fetch_ontologies(self) -> list[Ontology]:
    """Synchronous wrapper for fetch_ontologies.

    For async usage, use afetch_ontologies() instead.
    """
    # Use a temporary client for this operation to avoid event loop cleanup issues
    return asyncio.run(self._fetch_ontologies_with_cleanup())

init_dataset(dataset_name) async

Initialize a Fuseki dataset.

This method creates a new dataset in Fuseki if it doesn't already exist. It uses Fuseki's admin API to create the dataset with TDB2 storage.

Uses a temporary client to avoid event loop cleanup issues when called from different async contexts.

Parameters:

Name Type Description Default
dataset_name

Name of the dataset to create.

 required
Note

This method will not fail if the dataset already exists.

Source code in ontocast/tool/triple_manager/fuseki.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
async def init_dataset(self, dataset_name):
    """Initialize a Fuseki dataset.

    This method creates a new dataset in Fuseki if it doesn't already exist.
    It uses Fuseki's admin API to create the dataset with TDB2 storage.

    Uses a temporary client to avoid event loop cleanup issues when called
    from different async contexts.

    Args:
        dataset_name: Name of the dataset to create.

    Note:
        This method will not fail if the dataset already exists.
    """
    # Use a temporary client to avoid event loop cleanup issues
    async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
        fuseki_admin_url = f"{self.uri}/$/datasets"

        payload = {"dbName": dataset_name, "dbType": "tdb2"}

        headers = {"Content-Type": "application/x-www-form-urlencoded"}

        response = await client.post(
            fuseki_admin_url, data=payload, headers=headers
        )

        if response.status_code == 200 or response.status_code == 201:
            logger.info(f"Fuseki dataset '{dataset_name}' created successfully.")
        elif response.status_code == 409:
            logger.info(
                f"Fuseki status code: {response.status_code}; {response.text.strip()}"
            )
        else:
            logger.error(
                f"Failed to create dataset {dataset_name}. Status code: {response.status_code}"
            )
            logger.error(f"Response: {response.text.strip()}")

serialize(o, **kwargs)

Synchronous wrapper for serialize.

For async usage, use aserialize() instead.

Source code in ontocast/tool/triple_manager/fuseki.py 
732
733
734
735
736
737
def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
    """Synchronous wrapper for serialize.

    For async usage, use aserialize() instead.
    """
    return asyncio.run(self._serialize_with_cleanup(o, **kwargs))

serialize_graph(graph, **kwargs)

Synchronous wrapper for serialize_graph.

For async usage, use aserialize_graph() instead.

Source code in ontocast/tool/triple_manager/fuseki.py 
658
659
660
661
662
663
def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
    """Synchronous wrapper for serialize_graph.

    For async usage, use aserialize_graph() instead.
    """
    return asyncio.run(self._serialize_graph_with_cleanup(graph, **kwargs))

update_dataset(new_dataset) async

Update the dataset name for this manager.

This method allows changing the dataset without recreating the entire manager, which is useful for API requests that specify different datasets.

Parameters:

Name Type Description Default
new_dataset str

The new dataset name to use.

 required
Source code in ontocast/tool/triple_manager/fuseki.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
async def update_dataset(self, new_dataset: str) -> None:
    """Update the dataset name for this manager.

    This method allows changing the dataset without recreating the entire
    manager, which is useful for API requests that specify different datasets.

    Args:
        new_dataset: The new dataset name to use.
    """
    if not new_dataset:
        raise ValueError("Dataset name cannot be empty")

    self.dataset = new_dataset
    await self.init_dataset(self.dataset)
    logger.info(f"Updated Fuseki dataset to: {self.dataset}")

LLMTool

Bases: Tool

Tool for interacting with language models.

This class provides a unified interface for working with different language model providers (OpenAI, Ollama) through LangChain. It supports both synchronous and asynchronous operations.

Attributes:

Name Type Description
config LLMConfig

LLMConfig object containing all LLM settings.
cache Any

Cacher instance for caching LLM responses.
Source code in ontocast/tool/llm.py 
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
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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
class LLMTool(Tool):
    """Tool for interacting with language models.

    This class provides a unified interface for working with different language model
    providers (OpenAI, Ollama) through LangChain. It supports both synchronous and
    asynchronous operations.

    Attributes:
        config: LLMConfig object containing all LLM settings.
        cache: Cacher instance for caching LLM responses.
    """

    config: LLMConfig = Field(default_factory=LLMConfig)
    cache: Any = Field(default=None, exclude=True)
    budget_tracker: Any = Field(default=None, exclude=True)

    def __init__(
        self,
        cache: Cacher | None = None,
        budget_tracker: Any = None,
        **kwargs,
    ):
        """Initialize the LLM tool.

        Args:
            cache: Optional shared Cacher instance. If None, creates a new one.
            budget_tracker: Optional budget tracker instance for usage statistics.
            **kwargs: Additional keyword arguments passed to the parent class.
        """
        super().__init__(**kwargs)
        self._llm = None
        self.budget_tracker = budget_tracker

        # Initialize cache - use shared cacher or create new one
        if cache is not None:
            self.cache = ToolCacher(cache, "llm")
        else:
            # Fallback for backward compatibility
            shared_cache = Cacher()
            self.cache = ToolCacher(shared_cache, "llm")

    @classmethod
    def create(
        cls,
        config: LLMConfig,
        cache: Cacher | None = None,
        budget_tracker: Any = None,
        **kwargs,
    ):
        """Create a new LLM tool instance synchronously.

        Args:
            config: LLMConfig object containing LLM settings.
            cache: Optional shared Cacher instance.
            budget_tracker: Optional budget tracker instance for usage statistics.
            **kwargs: Additional keyword arguments for initialization.

        Returns:
            LLMTool: A new instance of the LLM tool.
        """
        return asyncio.run(
            cls.acreate(
                config=config, cache=cache, budget_tracker=budget_tracker, **kwargs
            )
        )

    @classmethod
    async def acreate(
        cls,
        config: LLMConfig,
        cache: Cacher | None = None,
        budget_tracker: Any = None,
        **kwargs,
    ):
        """Create a new LLM tool instance asynchronously.

        Args:
            config: LLMConfig object containing LLM settings.
            cache: Optional shared Cacher instance.
            budget_tracker: Optional budget tracker instance for usage statistics.
            **kwargs: Additional keyword arguments for initialization.

        Returns:
            LLMTool: A new instance of the LLM tool.
        """
        # Create and initialize the instance with the config
        self = cls(config=config, cache=cache, budget_tracker=budget_tracker, **kwargs)
        await self.setup()
        return self

    async def setup(self):
        """Set up the language model based on the configured provider.

        Raises:
            ValueError: If the provider is not supported.
        """
        if self.config.provider == LLMProvider.OPENAI:
            if self.config.model_name.startswith("gpt-5"):
                self.config.temperature = 1.0
                logger.warning(
                    f"Setting temperature to {self.config.temperature} for gpt-5 class "
                    f"model {self.config.model_name}"
                )
            self._llm = ChatOpenAI(
                model=self.config.model_name,  # type: ignore
                temperature=self.config.temperature,
                base_url=self.config.base_url,  # type: ignore
                api_key=(
                    SecretStr(self.config.api_key) if self.config.api_key else None
                ),  # type: ignore
            )
        elif self.config.provider == LLMProvider.OLLAMA:
            self._llm = ChatOllama(
                model=self.config.model_name,
                base_url=self.config.base_url,
                temperature=self.config.temperature,
            )
        else:
            raise ValueError(f"Unsupported provider: {self.config.provider}")

    @track_llm_usage
    async def __call__(self, *args: Any, **kwds: Any) -> Any:
        """Call the language model directly (asynchronous).

        Args:
            *args: Positional arguments passed to the LLM.
            **kwds: Keyword arguments passed to the LLM.

        Returns:
            Any: The LLM's response.
        """
        # Extract prompt from args (first argument is typically the prompt)
        prompt = args[0] if args else ""

        # Prepare configuration for caching
        config_dict = {
            "provider": self.config.provider,
            "model_name": self.config.model_name,
            "temperature": self.config.temperature,
            "base_url": self.config.base_url,
        }

        # Check cache first
        cached_response = self.cache.get(prompt, config=config_dict, **kwds)

        if cached_response is not None:
            prompt_str = self._prompt_to_string(prompt)
            logger.debug(f"Cache hit for __call__: {prompt_str[:50]}...")
            # Return a mock BaseMessage object with the cached content
            content = cached_response["content"]
            content_str = content if isinstance(content, str) else str(content)
            return AIMessage(content=content_str)

        # Generate new response
        prompt_str = self._prompt_to_string(prompt)
        logger.debug(f"Cache miss, calling LLM for __call__: {prompt_str[:50]}...")

        response = await self.llm.ainvoke(*args, **kwds)

        # Cache the response
        response_data = {
            "content": response.content,
            "prompt": self._prompt_to_string(prompt),
            "kwargs": kwds,
        }
        self.cache.set(prompt, response_data, config=config_dict, **kwds)

        return response

    @track_llm_usage
    async def acall(self, *args: Any, **kwds: Any) -> Any:
        """Call the language model directly (asynchronous).

        Args:
            *args: Positional arguments passed to the LLM.
            **kwds: Keyword arguments passed to the LLM.

        Returns:
            Any: The LLM's response.
        """
        # Extract prompt from args (first argument is typically the prompt)
        prompt = args[0] if args else ""

        # Prepare configuration for caching
        config_dict = {
            "provider": self.config.provider,
            "model_name": self.config.model_name,
            "temperature": self.config.temperature,
            "base_url": self.config.base_url,
        }

        # Check cache first
        cached_response = self.cache.get(prompt, config=config_dict, **kwds)

        if cached_response is not None:
            prompt_str = self._prompt_to_string(prompt)
            logger.debug(f"Cache hit for acall: {prompt_str[:50]}...")
            # Return a mock BaseMessage object with the cached content
            content = cached_response["content"]
            content_str = content if isinstance(content, str) else str(content)
            return AIMessage(content=content_str)

        # Generate new response
        prompt_str = self._prompt_to_string(prompt)
        logger.debug(f"Cache miss, calling LLM for acall: {prompt_str[:50]}...")

        response = await self.llm.ainvoke(*args, **kwds)

        # Cache the response
        response_data = {
            "content": response.content,
            "prompt": self._prompt_to_string(prompt),
            "kwargs": kwds,
        }
        self.cache.set(prompt, response_data, config=config_dict, **kwds)

        return response

    @property
    def llm(self) -> BaseChatModel:
        """Get the underlying language model instance.

        Returns:
            BaseChatModel: The configured language model.

        Raises:
            RuntimeError: If the LLM has not been properly initialized.
        """
        if self._llm is None:
            raise RuntimeError(
                "LLM resource not properly initialized. Call setup() first."
            )
        return self._llm

    def _prompt_to_string(self, prompt) -> str:
        """Convert various prompt types to string for caching.

        Args:
            prompt: The prompt object (string, StringPromptValue, etc.)

        Returns:
            str: String representation of the prompt.
        """
        if isinstance(prompt, str):
            return prompt
        elif hasattr(prompt, "to_string"):
            return prompt.to_string()
        elif hasattr(prompt, "text"):
            return prompt.text
        elif hasattr(prompt, "content"):
            return prompt.content
        else:
            return str(prompt)

    @track_llm_usage
    async def complete(self, prompt: str, **kwargs) -> Any:
        """Generate a completion for the given prompt.

        Args:
            prompt: The input prompt for generation.
            **kwargs: Additional keyword arguments for generation.

        Returns:
            Any: The generated completion.
        """
        # Prepare configuration for caching
        config_dict = {
            "provider": self.config.provider,
            "model_name": self.config.model_name,
            "temperature": self.config.temperature,
            "base_url": self.config.base_url,
        }

        # Check cache first
        cached_response = self.cache.get(prompt, config=config_dict, **kwargs)

        if cached_response is not None:
            logger.debug(f"Cache hit for prompt: {prompt[:50]}...")
            content = cached_response["content"]
            return content if isinstance(content, str) else str(content)

        # Generate new response
        logger.debug(f"Cache miss, calling LLM for prompt: {prompt[:50]}...")

        response = await self.llm.ainvoke(prompt, **kwargs)

        # Cache the response
        response_data = {
            "content": response.content,
            "prompt": self._prompt_to_string(prompt),
            "kwargs": kwargs,
        }
        self.cache.set(prompt, response_data, config=config_dict, **kwargs)

        return response.content

    @track_llm_usage
    async def extract(self, prompt: str, output_schema: Type[T], **kwargs) -> T:
        """Extract structured data from the prompt according to a schema.

        Args:
            prompt: The input prompt for extraction.
            output_schema: The Pydantic model class defining the output structure.
            **kwargs: Additional keyword arguments for extraction.

        Returns:
            T: The extracted data conforming to the output schema.
        """
        parser = PydanticOutputParser(pydantic_object=output_schema)
        format_instructions = parser.get_format_instructions()

        full_prompt = f"{prompt}\n\n{format_instructions}"

        # Prepare configuration for caching
        config_dict = {
            "provider": self.config.provider,
            "model_name": self.config.model_name,
            "temperature": self.config.temperature,
            "base_url": self.config.base_url,
            "output_schema": output_schema.__name__,
        }

        # Check cache first
        cached_response = self.cache.get(full_prompt, config=config_dict, **kwargs)

        if cached_response is not None:
            logger.debug(f"Cache hit for extraction: {prompt[:50]}...")
            # Parse the cached content
            content = cached_response["content"]
            if isinstance(content, str):
                return parser.parse(content)
            else:
                # Fallback: convert to string if it's not already
                return parser.parse(str(content))

        # Generate new response
        logger.debug(f"Cache miss, calling LLM for extraction: {prompt[:50]}...")

        response = await self.llm.ainvoke(full_prompt, **kwargs)

        # Cache the response
        response_data = {
            "content": response.content,
            "prompt": self._prompt_to_string(full_prompt),
            "output_schema": output_schema.__name__,
            "kwargs": kwargs,
        }
        self.cache.set(full_prompt, response_data, config=config_dict, **kwargs)

        content = response.content
        return parser.parse(content if isinstance(content, str) else str(content))

llm property

Get the underlying language model instance.

Returns:

Name Type Description
BaseChatModel BaseChatModel

The configured language model.

Raises:

Type Description
RuntimeError

If the LLM has not been properly initialized.

__call__(*args, **kwds) async

Call the language model directly (asynchronous).

Parameters:

Name Type Description Default
*args Any

Positional arguments passed to the LLM.

 ()
**kwds Any

Keyword arguments passed to the LLM.

 {}

Returns:

Name Type Description
Any Any

The LLM's response.
Source code in ontocast/tool/llm.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
@track_llm_usage
async def __call__(self, *args: Any, **kwds: Any) -> Any:
    """Call the language model directly (asynchronous).

    Args:
        *args: Positional arguments passed to the LLM.
        **kwds: Keyword arguments passed to the LLM.

    Returns:
        Any: The LLM's response.
    """
    # Extract prompt from args (first argument is typically the prompt)
    prompt = args[0] if args else ""

    # Prepare configuration for caching
    config_dict = {
        "provider": self.config.provider,
        "model_name": self.config.model_name,
        "temperature": self.config.temperature,
        "base_url": self.config.base_url,
    }

    # Check cache first
    cached_response = self.cache.get(prompt, config=config_dict, **kwds)

    if cached_response is not None:
        prompt_str = self._prompt_to_string(prompt)
        logger.debug(f"Cache hit for __call__: {prompt_str[:50]}...")
        # Return a mock BaseMessage object with the cached content
        content = cached_response["content"]
        content_str = content if isinstance(content, str) else str(content)
        return AIMessage(content=content_str)

    # Generate new response
    prompt_str = self._prompt_to_string(prompt)
    logger.debug(f"Cache miss, calling LLM for __call__: {prompt_str[:50]}...")

    response = await self.llm.ainvoke(*args, **kwds)

    # Cache the response
    response_data = {
        "content": response.content,
        "prompt": self._prompt_to_string(prompt),
        "kwargs": kwds,
    }
    self.cache.set(prompt, response_data, config=config_dict, **kwds)

    return response

__init__(cache=None, budget_tracker=None, **kwargs)

Initialize the LLM tool.

Parameters:

Name Type Description Default
cache Cacher | None

Optional shared Cacher instance. If None, creates a new one.

 None
budget_tracker Any

Optional budget tracker instance for usage statistics.

 None
**kwargs

Additional keyword arguments passed to the parent class.

 {}
Source code in ontocast/tool/llm.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def __init__(
    self,
    cache: Cacher | None = None,
    budget_tracker: Any = None,
    **kwargs,
):
    """Initialize the LLM tool.

    Args:
        cache: Optional shared Cacher instance. If None, creates a new one.
        budget_tracker: Optional budget tracker instance for usage statistics.
        **kwargs: Additional keyword arguments passed to the parent class.
    """
    super().__init__(**kwargs)
    self._llm = None
    self.budget_tracker = budget_tracker

    # Initialize cache - use shared cacher or create new one
    if cache is not None:
        self.cache = ToolCacher(cache, "llm")
    else:
        # Fallback for backward compatibility
        shared_cache = Cacher()
        self.cache = ToolCacher(shared_cache, "llm")

acall(*args, **kwds) async

Call the language model directly (asynchronous).

Parameters:

Name Type Description Default
*args Any

Positional arguments passed to the LLM.

 ()
**kwds Any

Keyword arguments passed to the LLM.

 {}

Returns:

Name Type Description
Any Any

The LLM's response.
Source code in ontocast/tool/llm.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
@track_llm_usage
async def acall(self, *args: Any, **kwds: Any) -> Any:
    """Call the language model directly (asynchronous).

    Args:
        *args: Positional arguments passed to the LLM.
        **kwds: Keyword arguments passed to the LLM.

    Returns:
        Any: The LLM's response.
    """
    # Extract prompt from args (first argument is typically the prompt)
    prompt = args[0] if args else ""

    # Prepare configuration for caching
    config_dict = {
        "provider": self.config.provider,
        "model_name": self.config.model_name,
        "temperature": self.config.temperature,
        "base_url": self.config.base_url,
    }

    # Check cache first
    cached_response = self.cache.get(prompt, config=config_dict, **kwds)

    if cached_response is not None:
        prompt_str = self._prompt_to_string(prompt)
        logger.debug(f"Cache hit for acall: {prompt_str[:50]}...")
        # Return a mock BaseMessage object with the cached content
        content = cached_response["content"]
        content_str = content if isinstance(content, str) else str(content)
        return AIMessage(content=content_str)

    # Generate new response
    prompt_str = self._prompt_to_string(prompt)
    logger.debug(f"Cache miss, calling LLM for acall: {prompt_str[:50]}...")

    response = await self.llm.ainvoke(*args, **kwds)

    # Cache the response
    response_data = {
        "content": response.content,
        "prompt": self._prompt_to_string(prompt),
        "kwargs": kwds,
    }
    self.cache.set(prompt, response_data, config=config_dict, **kwds)

    return response

acreate(config, cache=None, budget_tracker=None, **kwargs) async classmethod

Create a new LLM tool instance asynchronously.

Parameters:

Name Type Description Default
config LLMConfig

LLMConfig object containing LLM settings.

 required
cache Cacher | None

Optional shared Cacher instance.

 None
budget_tracker Any

Optional budget tracker instance for usage statistics.

 None
**kwargs

Additional keyword arguments for initialization.

 {}

Returns:

Name Type Description
LLMTool

A new instance of the LLM tool.
Source code in ontocast/tool/llm.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@classmethod
async def acreate(
    cls,
    config: LLMConfig,
    cache: Cacher | None = None,
    budget_tracker: Any = None,
    **kwargs,
):
    """Create a new LLM tool instance asynchronously.

    Args:
        config: LLMConfig object containing LLM settings.
        cache: Optional shared Cacher instance.
        budget_tracker: Optional budget tracker instance for usage statistics.
        **kwargs: Additional keyword arguments for initialization.

    Returns:
        LLMTool: A new instance of the LLM tool.
    """
    # Create and initialize the instance with the config
    self = cls(config=config, cache=cache, budget_tracker=budget_tracker, **kwargs)
    await self.setup()
    return self

complete(prompt, **kwargs) async

Generate a completion for the given prompt.

Parameters:

Name Type Description Default
prompt str

The input prompt for generation.

 required
**kwargs

Additional keyword arguments for generation.

 {}

Returns:

Name Type Description
Any Any

The generated completion.
Source code in ontocast/tool/llm.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
@track_llm_usage
async def complete(self, prompt: str, **kwargs) -> Any:
    """Generate a completion for the given prompt.

    Args:
        prompt: The input prompt for generation.
        **kwargs: Additional keyword arguments for generation.

    Returns:
        Any: The generated completion.
    """
    # Prepare configuration for caching
    config_dict = {
        "provider": self.config.provider,
        "model_name": self.config.model_name,
        "temperature": self.config.temperature,
        "base_url": self.config.base_url,
    }

    # Check cache first
    cached_response = self.cache.get(prompt, config=config_dict, **kwargs)

    if cached_response is not None:
        logger.debug(f"Cache hit for prompt: {prompt[:50]}...")
        content = cached_response["content"]
        return content if isinstance(content, str) else str(content)

    # Generate new response
    logger.debug(f"Cache miss, calling LLM for prompt: {prompt[:50]}...")

    response = await self.llm.ainvoke(prompt, **kwargs)

    # Cache the response
    response_data = {
        "content": response.content,
        "prompt": self._prompt_to_string(prompt),
        "kwargs": kwargs,
    }
    self.cache.set(prompt, response_data, config=config_dict, **kwargs)

    return response.content

create(config, cache=None, budget_tracker=None, **kwargs) classmethod

Create a new LLM tool instance synchronously.

Parameters:

Name Type Description Default
config LLMConfig

LLMConfig object containing LLM settings.

 required
cache Cacher | None

Optional shared Cacher instance.

 None
budget_tracker Any

Optional budget tracker instance for usage statistics.

 None
**kwargs

Additional keyword arguments for initialization.

 {}

Returns:

Name Type Description
LLMTool

A new instance of the LLM tool.
Source code in ontocast/tool/llm.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@classmethod
def create(
    cls,
    config: LLMConfig,
    cache: Cacher | None = None,
    budget_tracker: Any = None,
    **kwargs,
):
    """Create a new LLM tool instance synchronously.

    Args:
        config: LLMConfig object containing LLM settings.
        cache: Optional shared Cacher instance.
        budget_tracker: Optional budget tracker instance for usage statistics.
        **kwargs: Additional keyword arguments for initialization.

    Returns:
        LLMTool: A new instance of the LLM tool.
    """
    return asyncio.run(
        cls.acreate(
            config=config, cache=cache, budget_tracker=budget_tracker, **kwargs
        )
    )

extract(prompt, output_schema, **kwargs) async

Extract structured data from the prompt according to a schema.

Parameters:

Name Type Description Default
prompt str

The input prompt for extraction.

 required
output_schema Type[T]

The Pydantic model class defining the output structure.

 required
**kwargs

Additional keyword arguments for extraction.

 {}

Returns:

Name Type Description
T T

The extracted data conforming to the output schema.
Source code in ontocast/tool/llm.py 
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
@track_llm_usage
async def extract(self, prompt: str, output_schema: Type[T], **kwargs) -> T:
    """Extract structured data from the prompt according to a schema.

    Args:
        prompt: The input prompt for extraction.
        output_schema: The Pydantic model class defining the output structure.
        **kwargs: Additional keyword arguments for extraction.

    Returns:
        T: The extracted data conforming to the output schema.
    """
    parser = PydanticOutputParser(pydantic_object=output_schema)
    format_instructions = parser.get_format_instructions()

    full_prompt = f"{prompt}\n\n{format_instructions}"

    # Prepare configuration for caching
    config_dict = {
        "provider": self.config.provider,
        "model_name": self.config.model_name,
        "temperature": self.config.temperature,
        "base_url": self.config.base_url,
        "output_schema": output_schema.__name__,
    }

    # Check cache first
    cached_response = self.cache.get(full_prompt, config=config_dict, **kwargs)

    if cached_response is not None:
        logger.debug(f"Cache hit for extraction: {prompt[:50]}...")
        # Parse the cached content
        content = cached_response["content"]
        if isinstance(content, str):
            return parser.parse(content)
        else:
            # Fallback: convert to string if it's not already
            return parser.parse(str(content))

    # Generate new response
    logger.debug(f"Cache miss, calling LLM for extraction: {prompt[:50]}...")

    response = await self.llm.ainvoke(full_prompt, **kwargs)

    # Cache the response
    response_data = {
        "content": response.content,
        "prompt": self._prompt_to_string(full_prompt),
        "output_schema": output_schema.__name__,
        "kwargs": kwargs,
    }
    self.cache.set(full_prompt, response_data, config=config_dict, **kwargs)

    content = response.content
    return parser.parse(content if isinstance(content, str) else str(content))

setup() async

Set up the language model based on the configured provider.

Raises:

Type Description
ValueError

If the provider is not supported.
Source code in ontocast/tool/llm.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
async def setup(self):
    """Set up the language model based on the configured provider.

    Raises:
        ValueError: If the provider is not supported.
    """
    if self.config.provider == LLMProvider.OPENAI:
        if self.config.model_name.startswith("gpt-5"):
            self.config.temperature = 1.0
            logger.warning(
                f"Setting temperature to {self.config.temperature} for gpt-5 class "
                f"model {self.config.model_name}"
            )
        self._llm = ChatOpenAI(
            model=self.config.model_name,  # type: ignore
            temperature=self.config.temperature,
            base_url=self.config.base_url,  # type: ignore
            api_key=(
                SecretStr(self.config.api_key) if self.config.api_key else None
            ),  # type: ignore
        )
    elif self.config.provider == LLMProvider.OLLAMA:
        self._llm = ChatOllama(
            model=self.config.model_name,
            base_url=self.config.base_url,
            temperature=self.config.temperature,
        )
    else:
        raise ValueError(f"Unsupported provider: {self.config.provider}")

Neo4jTripleStoreManager

Bases: TripleStoreManagerWithAuth

Neo4j-based triple store manager using n10s (neosemantics) plugin.

This implementation handles RDF data more faithfully by using both the n10s property graph representation and raw RDF triple storage for accurate reconstruction. It provides comprehensive ontology management with namespace-based organization.

The manager uses Neo4j's n10s plugin for RDF operations, including: - RDF import and export via n10s - Ontology metadata storage and retrieval - Namespace-based ontology organization - Faithful RDF graph reconstruction

Attributes:

Name Type Description
_driver Any

Private Neo4j driver instance.
Source code in ontocast/tool/triple_manager/neo4j.py 
 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
144
145
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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
class Neo4jTripleStoreManager(TripleStoreManagerWithAuth):
    """Neo4j-based triple store manager using n10s (neosemantics) plugin.

    This implementation handles RDF data more faithfully by using both the n10s
    property graph representation and raw RDF triple storage for accurate reconstruction.
    It provides comprehensive ontology management with namespace-based organization.

    The manager uses Neo4j's n10s plugin for RDF operations, including:
    - RDF import and export via n10s
    - Ontology metadata storage and retrieval
    - Namespace-based ontology organization
    - Faithful RDF graph reconstruction

    Attributes:
        _driver: Private Neo4j driver instance.
    """

    _driver: Any = None  # private attribute, not a pydantic field

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

        This method sets up the connection to Neo4j, initializes the n10s
        plugin configuration, and creates necessary constraints and indexes.
        The database is NOT cleaned on initialization.

        Args:
            uri: Neo4j connection URI (e.g., "bolt://localhost:7687").
            auth: Authentication tuple (username, password) or string in "user/password" format.
            **kwargs: Additional keyword arguments passed to the parent class.

        Raises:
            ImportError: If the neo4j Python driver is not installed.

        Example:
            >>> manager = Neo4jTripleStoreManager(
            ...     uri="bolt://localhost:7687",
            ...     auth="neo4j/password"
            ... )
            >>> # To clean the database, use the clean() method explicitly:
            >>> await manager.clean()
        """
        super().__init__(
            uri=uri, auth=auth, env_uri="NEO4J_URI", env_auth="NEO4J_AUTH", **kwargs
        )
        if GraphDatabase is None:
            raise ImportError("neo4j Python driver is not installed.")
        if self.uri is None:
            raise ValueError("Neo4j URI is required but not provided.")
        self._driver = GraphDatabase.driver(self.uri, auth=self.auth)

        # Type assertion: we know _driver is not None after initialization
        assert self._driver is not None

        with self._driver.session() as session:
            # Initialize n10s configuration
            self._init_n10s_config(session)

            # Create constraints and indexes
            self._create_constraints_and_indexes(session)

    async def clean(self, dataset: str | None = None) -> None:
        """Clean/flush all data from the Neo4j database.

        This method deletes all nodes and relationships from the Neo4j database,
        effectively clearing all stored data.

        Args:
            dataset: Optional dataset parameter (ignored for Neo4j, which doesn't
                support datasets). Included for interface compatibility.

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

        Raises:
            Exception: If the cleanup operation fails.
        """
        if dataset is not None:
            logger.warning(
                f"Dataset parameter '{dataset}' ignored for Neo4j (datasets not supported)"
            )

        if self._driver is None:
            raise ValueError("Neo4j driver is not initialized")

        with self._driver.session() as session:
            try:
                session.run("MATCH (n) DETACH DELETE n")
                logger.info("Neo4j database cleaned (all nodes deleted)")
            except Exception as e:
                logger.error(f"Neo4j cleanup failed: {e}")
                raise

    def _init_n10s_config(self, session):
        """Initialize n10s configuration with better RDF handling.

        This method configures the n10s plugin for optimal RDF handling.
        It sets up the configuration to preserve vocabulary URIs, handle
        multivalued properties, and maintain RDF types as nodes.

        Args:
            session: Neo4j session for executing configuration commands.
        """
        try:
            # Check if already configured
            result = session.run("CALL n10s.graphconfig.show()")
            if result.single():
                logger.debug("n10s already configured")
        except:
            pass

        try:
            session.run("""
                CALL n10s.graphconfig.init({
                    handleVocabUris: "KEEP",
                    handleMultival: "OVERWRITE",
                    typesToLabels: false,
                    keepLangTag: false,
                    keepCustomDataTypes: true,
                    handleRDFTypes: "NODES"
                })
            """)
            logger.debug("n10s configuration initialized")
        except Exception as e:
            logger.warning(f"n10s configuration failed: {e}")

    def _create_constraints_and_indexes(self, session):
        """Create necessary constraints and indexes for optimal performance.

        This method creates Neo4j constraints and indexes that are needed
        for efficient ontology operations and data integrity.

        Args:
            session: Neo4j session for executing constraint/index creation commands.
        """
        constraints = [
            "CREATE CONSTRAINT n10s_unique_uri IF NOT EXISTS FOR (r:Resource) REQUIRE r.uri IS UNIQUE",
            "CREATE CONSTRAINT ontology_iri_unique IF NOT EXISTS FOR (o:Ontology) REQUIRE o.uri IS UNIQUE",
            "CREATE INDEX namespace_prefix IF NOT EXISTS FOR (ns:Namespace) ON (ns.prefix)",
        ]

        for constraint in constraints:
            try:
                session.run(constraint)
                logger.debug(f"Created constraint/index: {constraint.split()[-1]}")
            except Exception as e:
                logger.debug(f"Constraint/index creation (might already exist): {e}")

    def _extract_namespace_prefix(self, uri: str) -> tuple[str, str]:
        """Extract namespace and local name from URI.

        This method parses a URI to extract the namespace and local name
        using common separators (#, /, :).

        Args:
            uri: The URI to parse.

        Returns:
            tuple[str, str]: A tuple of (namespace, local_name).

        Example:
            >>> manager._extract_namespace_prefix("http://example.org/onto#Class")
            ("http://example.org/onto#", "Class")
        """
        common_separators = ["#", "/", ":"]
        for sep in common_separators:
            if sep in uri:
                parts = uri.rsplit(sep, 1)
                if len(parts) == 2:
                    return parts[0] + sep, parts[1]
        return uri, ""

    def _get_ontology_namespaces(self, session) -> dict:
        """Get all known ontology namespaces from the database.

        This method queries the Neo4j database to retrieve all known
        namespace prefixes and their corresponding URIs.

        Args:
            session: Neo4j session for executing queries.

        Returns:
            dict: Dictionary mapping namespace prefixes to URIs.
        """
        result = session.run("""
            MATCH (ns:Namespace)
            RETURN ns.prefix as prefix, ns.uri as uri
            UNION
            MATCH (o:Ontology)
            RETURN null as prefix, o.uri as uri
        """)

        namespaces = {}
        for record in result:
            uri = record.get("uri")
            prefix = record.get("prefix")
            if uri:
                if prefix:
                    namespaces[prefix] = uri
                else:
                    # Extract potential namespace from ontology URI
                    ns, _ = self._extract_namespace_prefix(uri)
                    if ns != uri:  # Only if we actually found a namespace
                        namespaces[ns] = ns

        return namespaces

    def fetch_ontologies(self) -> list[Ontology]:
        """Fetch ontologies from Neo4j with faithful RDF reconstruction.

        This method retrieves all ontologies from Neo4j and reconstructs
        their RDF graphs faithfully. It uses a multi-step process:

        1. Identifies distinct ontologies by their namespace URIs
        2. Fetches all entities belonging to each ontology
        3. Reconstructs the RDF graph faithfully using stored triples when available
        4. Falls back to n10s property graph conversion when needed

        Returns:
            list[Ontology]: List of all ontologies found in the database.

        Example:
            >>> ontologies = manager.fetch_ontologies()
            >>> for onto in ontologies:
            ...     print(f"Found ontology: {onto.iri}")
        """
        ontologies = []

        # Type assertion: we know _driver is not None after initialization
        assert self._driver is not None
        with self._driver.session() as session:
            try:
                # First, try to get explicitly stored ontology metadata
                ontology_iris = self._fetch_ontology_iris(session)

                if ontology_iris:
                    for ont_iri in ontology_iris:
                        ontology = self._reconstruct_ontology_from_metadata(
                            session, ont_iri
                        )
                        if ontology:
                            ontologies.append(ontology)

            except Exception as e:
                logger.error(f"Error in fetch_ontologies: {e}")

        logger.info(f"Successfully loaded {len(ontologies)} ontologies")
        return ontologies

    def _fetch_ontology_iris(self, session) -> list[str]:
        """Fetch explicit ontology metadata from Neo4j.

        This method queries Neo4j to find all entities that are explicitly
        typed as owl:Ontology.

        Args:
            session: Neo4j session for executing queries.

        Returns:
            list[str]: List of ontology IRIs found in the database.
        """
        result = session.run(f"""
            MATCH (o)-[:`{str(RDF.type)}`]->(t:Resource {{ uri: "{str(OWL.Ontology)}" }})
            WHERE o.uri IS NOT NULL
            RETURN
              o.uri AS iri
        """)

        iris = []
        for record in result:
            iri = record.get("iri", None)
            iris += [iri]
        iris = [iri for iri in iris if iri is not None]
        return iris

    def _reconstruct_ontology_from_metadata(self, session, iri) -> Ontology | None:
        """Reconstruct an ontology from its metadata and related entities.

        This method takes an ontology IRI and reconstructs the complete
        ontology by fetching all related entities from the namespace.

        Args:
            session: Neo4j session for executing queries.
            iri: The ontology IRI to reconstruct.

        Returns:
            Ontology | None: The reconstructed ontology, or None if failed.
        """
        namespace_uri, _ = self._extract_namespace_prefix(iri)

        logger.debug(f"Reconstructing ontology: {iri} with namespace: {namespace_uri}")

        # Fallback to n10s export for this namespace
        graph = self._export_namespace_via_n10s(session, namespace_uri)
        if graph and len(graph) > 0:
            return self._create_ontology_object(iri, iri, graph)

    def _export_namespace_via_n10s(
        self, session, namespace_uri: str
    ) -> RDFGraph | None:
        """Export entities belonging to a namespace using n10s.

        This method uses Neo4j's n10s plugin to export all entities
        belonging to a specific namespace as RDF triples.

        Args:
            session: Neo4j session for executing queries.
            namespace_uri: The namespace URI to export.

        Returns:
            RDFGraph | None: The exported RDF graph, or None if failed.
        """
        try:
            result = session.run(
                f"""
                CALL n10s.rdf.export.cypher(
                    'MATCH (n)-[r]->(m) WHERE n.uri STARTS WITH "{namespace_uri}" RETURN n,r,m',
                    {{format: 'Turtle'}}
                )
                YIELD subject, predicate, object, isLiteral, literalType, literalLang
                RETURN subject, predicate, object, isLiteral, literalType, literalLang
                """
            )

            # Process into Turtle format
            turtle_lines = []

            for record in result:
                subj = record["subject"]
                pred = record["predicate"]
                obj = record["object"]
                is_literal = record["isLiteral"]
                literal_type = record["literalType"]
                literal_lang = record["literalLang"]

                # Format object
                if is_literal:
                    # Escape special characters in literals
                    obj = obj.replace('"', r"\"")
                    obj_str = f'"{obj}"'

                    # Add datatype or language tag if present
                    if literal_lang:
                        obj_str += f"@{literal_lang}"
                    elif literal_type:
                        obj_str += f"^^<{literal_type}>"
                else:
                    obj_str = f"<{obj}>"

                # Format triple
                turtle_lines.append(f"<{subj}> <{pred}> {obj_str} .")

            # Combine into single string
            turtle_string = "\n".join(turtle_lines)

            if turtle_string.strip():
                graph = RDFGraph()
                graph.parse(data=turtle_string, format="turtle")
                logger.debug(
                    f"Exported {len(graph)} triples via n10s for namespace {namespace_uri}"
                )
                return graph
            return None

        except Exception as e:
            logger.debug(
                f"Failed to export via n10s for namespace {namespace_uri}: {e}"
            )

        return None

    def _create_ontology_object(
        self, iri: str, metadata: dict, graph: RDFGraph
    ) -> Ontology:
        """Create an Ontology object from IRI, metadata, and graph.

        Args:
            iri: The ontology IRI.
            metadata: Metadata dictionary (currently unused, kept for compatibility).
            graph: The RDF graph containing the ontology data.

        Returns:
            Ontology: The created ontology object.
        """
        ontology_id = derive_ontology_id(iri)
        return Ontology(graph=graph, iri=iri, ontology_id=ontology_id)

    def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
        """Serialize an RDF graph to Neo4j with both n10s and raw triple storage.

        This method stores the given RDF graph in Neo4j using the n10s plugin
        for RDF import. The data is stored as RDF triples that can be faithfully
        reconstructed later.

        Args:
            graph: The RDF graph to store.
            **kwargs: Additional parameters (not used by Neo4j implementation).

        Returns:
            Any: The result summary from n10s import operation.
        """
        # Convert to RDFGraph if needed
        if not isinstance(graph, RDFGraph):
            rdf_graph = RDFGraph()
            for triple in graph:
                rdf_graph.add(triple)
            for prefix, namespace in graph.namespaces():
                rdf_graph.bind(prefix, namespace)
            graph = rdf_graph

        turtle_data = graph.serialize(format="turtle")

        # Type assertion: we know _driver is not None after initialization
        assert self._driver is not None
        with self._driver.session() as session:
            # Store via n10s for graph queries
            result = session.run(
                "CALL n10s.rdf.import.inline($ttl, 'Turtle')", ttl=turtle_data
            )
            summary = result.single()

        return summary

    def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
        """Serialize an Ontology or RDFGraph to Neo4j with both n10s and raw triple storage.

        This method stores the given Ontology or RDFGraph in Neo4j using the n10s plugin
        for RDF import. The data is stored as RDF triples that can be faithfully
        reconstructed later.

        Args:
            o: Ontology or RDFGraph object to store.
            **kwargs: Additional keyword arguments (not used by Neo4j implementation).

        Returns:
            Any: The result summary from n10s import operation.
        """
        if isinstance(o, Ontology):
            graph = o.graph
        elif isinstance(o, RDFGraph):
            graph = o
        else:
            raise TypeError(f"unsupported obj of type {type(o)} received")

        return self.serialize_graph(graph)

    def close(self):
        """Close the Neo4j driver connection.

        This method should be called when the manager is no longer needed
        to properly close the database connection and free resources.
        """
        if self._driver:
            self._driver.close()

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

Initialize the Neo4j triple store manager.

This method sets up the connection to Neo4j, initializes the n10s plugin configuration, and creates necessary constraints and indexes. The database is NOT cleaned on initialization.

Parameters:

Name Type Description Default
uri

Neo4j connection URI (e.g., "bolt://localhost:7687").

 None
auth

Authentication tuple (username, password) or string in "user/password" format.

 None
**kwargs

Additional keyword arguments passed to the parent class.

 {}

Raises:

Type Description
ImportError

If the neo4j Python driver is not installed.
Example

manager = Neo4jTripleStoreManager( ... uri="bolt://localhost:7687", ... auth="neo4j/password" ... )

To clean the database, use the clean() method explicitly:

await manager.clean()

Source code in ontocast/tool/triple_manager/neo4j.py 
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
def __init__(self, uri=None, auth=None, **kwargs):
    """Initialize the Neo4j triple store manager.

    This method sets up the connection to Neo4j, initializes the n10s
    plugin configuration, and creates necessary constraints and indexes.
    The database is NOT cleaned on initialization.

    Args:
        uri: Neo4j connection URI (e.g., "bolt://localhost:7687").
        auth: Authentication tuple (username, password) or string in "user/password" format.
        **kwargs: Additional keyword arguments passed to the parent class.

    Raises:
        ImportError: If the neo4j Python driver is not installed.

    Example:
        >>> manager = Neo4jTripleStoreManager(
        ...     uri="bolt://localhost:7687",
        ...     auth="neo4j/password"
        ... )
        >>> # To clean the database, use the clean() method explicitly:
        >>> await manager.clean()
    """
    super().__init__(
        uri=uri, auth=auth, env_uri="NEO4J_URI", env_auth="NEO4J_AUTH", **kwargs
    )
    if GraphDatabase is None:
        raise ImportError("neo4j Python driver is not installed.")
    if self.uri is None:
        raise ValueError("Neo4j URI is required but not provided.")
    self._driver = GraphDatabase.driver(self.uri, auth=self.auth)

    # Type assertion: we know _driver is not None after initialization
    assert self._driver is not None

    with self._driver.session() as session:
        # Initialize n10s configuration
        self._init_n10s_config(session)

        # Create constraints and indexes
        self._create_constraints_and_indexes(session)

clean(dataset=None) async

Clean/flush all data from the Neo4j database.

This method deletes all nodes and relationships from the Neo4j database, effectively clearing all stored data.

Parameters:

Name Type Description Default
dataset str | None

Optional dataset parameter (ignored for Neo4j, which doesn't support datasets). Included for interface compatibility.

 None

Raises:

Type Description
Exception

If the cleanup operation fails.
Source code in ontocast/tool/triple_manager/neo4j.py 
 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
async def clean(self, dataset: str | None = None) -> None:
    """Clean/flush all data from the Neo4j database.

    This method deletes all nodes and relationships from the Neo4j database,
    effectively clearing all stored data.

    Args:
        dataset: Optional dataset parameter (ignored for Neo4j, which doesn't
            support datasets). Included for interface compatibility.

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

    Raises:
        Exception: If the cleanup operation fails.
    """
    if dataset is not None:
        logger.warning(
            f"Dataset parameter '{dataset}' ignored for Neo4j (datasets not supported)"
        )

    if self._driver is None:
        raise ValueError("Neo4j driver is not initialized")

    with self._driver.session() as session:
        try:
            session.run("MATCH (n) DETACH DELETE n")
            logger.info("Neo4j database cleaned (all nodes deleted)")
        except Exception as e:
            logger.error(f"Neo4j cleanup failed: {e}")
            raise

close()

Close the Neo4j driver connection.

This method should be called when the manager is no longer needed to properly close the database connection and free resources.

Source code in ontocast/tool/triple_manager/neo4j.py 
469
470
471
472
473
474
475
476
def close(self):
    """Close the Neo4j driver connection.

    This method should be called when the manager is no longer needed
    to properly close the database connection and free resources.
    """
    if self._driver:
        self._driver.close()

fetch_ontologies()

Fetch ontologies from Neo4j with faithful RDF reconstruction.

This method retrieves all ontologies from Neo4j and reconstructs their RDF graphs faithfully. It uses a multi-step process:

  1. Identifies distinct ontologies by their namespace URIs
  2. Fetches all entities belonging to each ontology
  3. Reconstructs the RDF graph faithfully using stored triples when available
  4. Falls back to n10s property graph conversion when needed

Returns:

Type Description
list[Ontology]

list[Ontology]: List of all ontologies found in the database.
Example

ontologies = manager.fetch_ontologies() for onto in ontologies: ... print(f"Found ontology: {onto.iri}")

Source code in ontocast/tool/triple_manager/neo4j.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def fetch_ontologies(self) -> list[Ontology]:
    """Fetch ontologies from Neo4j with faithful RDF reconstruction.

    This method retrieves all ontologies from Neo4j and reconstructs
    their RDF graphs faithfully. It uses a multi-step process:

    1. Identifies distinct ontologies by their namespace URIs
    2. Fetches all entities belonging to each ontology
    3. Reconstructs the RDF graph faithfully using stored triples when available
    4. Falls back to n10s property graph conversion when needed

    Returns:
        list[Ontology]: List of all ontologies found in the database.

    Example:
        >>> ontologies = manager.fetch_ontologies()
        >>> for onto in ontologies:
        ...     print(f"Found ontology: {onto.iri}")
    """
    ontologies = []

    # Type assertion: we know _driver is not None after initialization
    assert self._driver is not None
    with self._driver.session() as session:
        try:
            # First, try to get explicitly stored ontology metadata
            ontology_iris = self._fetch_ontology_iris(session)

            if ontology_iris:
                for ont_iri in ontology_iris:
                    ontology = self._reconstruct_ontology_from_metadata(
                        session, ont_iri
                    )
                    if ontology:
                        ontologies.append(ontology)

        except Exception as e:
            logger.error(f"Error in fetch_ontologies: {e}")

    logger.info(f"Successfully loaded {len(ontologies)} ontologies")
    return ontologies

serialize(o, **kwargs)

Serialize an Ontology or RDFGraph to Neo4j with both n10s and raw triple storage.

This method stores the given Ontology or RDFGraph in Neo4j using the n10s plugin for RDF import. The data is stored as RDF triples that can be faithfully reconstructed later.

Parameters:

Name Type Description Default
o Ontology | RDFGraph

Ontology or RDFGraph object to store.

 required
**kwargs

Additional keyword arguments (not used by Neo4j implementation).

 {}

Returns:

Name Type Description
Any bool | None

The result summary from n10s import operation.
Source code in ontocast/tool/triple_manager/neo4j.py 
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool | None:
    """Serialize an Ontology or RDFGraph to Neo4j with both n10s and raw triple storage.

    This method stores the given Ontology or RDFGraph in Neo4j using the n10s plugin
    for RDF import. The data is stored as RDF triples that can be faithfully
    reconstructed later.

    Args:
        o: Ontology or RDFGraph object to store.
        **kwargs: Additional keyword arguments (not used by Neo4j implementation).

    Returns:
        Any: The result summary from n10s import operation.
    """
    if isinstance(o, Ontology):
        graph = o.graph
    elif isinstance(o, RDFGraph):
        graph = o
    else:
        raise TypeError(f"unsupported obj of type {type(o)} received")

    return self.serialize_graph(graph)

serialize_graph(graph, **kwargs)

Serialize an RDF graph to Neo4j with both n10s and raw triple storage.

This method stores the given RDF graph in Neo4j using the n10s plugin for RDF import. The data is stored as RDF triples that can be faithfully reconstructed later.

Parameters:

Name Type Description Default
graph Graph

The RDF graph to store.

 required
**kwargs

Additional parameters (not used by Neo4j implementation).

 {}

Returns:

Name Type Description
Any bool | None

The result summary from n10s import operation.
Source code in ontocast/tool/triple_manager/neo4j.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def serialize_graph(self, graph: Graph, **kwargs) -> bool | None:
    """Serialize an RDF graph to Neo4j with both n10s and raw triple storage.

    This method stores the given RDF graph in Neo4j using the n10s plugin
    for RDF import. The data is stored as RDF triples that can be faithfully
    reconstructed later.

    Args:
        graph: The RDF graph to store.
        **kwargs: Additional parameters (not used by Neo4j implementation).

    Returns:
        Any: The result summary from n10s import operation.
    """
    # Convert to RDFGraph if needed
    if not isinstance(graph, RDFGraph):
        rdf_graph = RDFGraph()
        for triple in graph:
            rdf_graph.add(triple)
        for prefix, namespace in graph.namespaces():
            rdf_graph.bind(prefix, namespace)
        graph = rdf_graph

    turtle_data = graph.serialize(format="turtle")

    # Type assertion: we know _driver is not None after initialization
    assert self._driver is not None
    with self._driver.session() as session:
        # Store via n10s for graph queries
        result = session.run(
            "CALL n10s.rdf.import.inline($ttl, 'Turtle')", ttl=turtle_data
        )
        summary = result.single()

    return summary

OntologyManager

Bases: Tool

Manager for handling multiple ontologies with version tracking.

This class provides functionality for managing a collection of ontologies, tracking version lineage using hash-based identifiers. For each IRI, it maintains a tree/graph of all versions identified by their hashes.

Attributes:

Name Type Description
ontology_versions dict[str, list[Ontology]]

Dictionary mapping IRI to list of all ontology versions (identified by hash). Each IRI can have multiple versions forming a lineage tree.
Source code in ontocast/tool/ontology_manager.py 
 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
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
144
145
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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
class OntologyManager(Tool):
    """Manager for handling multiple ontologies with version tracking.

    This class provides functionality for managing a collection of ontologies,
    tracking version lineage using hash-based identifiers. For each IRI,
    it maintains a tree/graph of all versions identified by their hashes.

    Attributes:
        ontology_versions: Dictionary mapping IRI to list of all
            ontology versions (identified by hash). Each IRI can have
            multiple versions forming a lineage tree.
    """

    ontology_versions: dict[str, list[Ontology]] = Field(default_factory=dict)

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

        Args:
            **kwargs: Additional keyword arguments passed to the parent class.
        """
        super().__init__(**kwargs)
        # Cache dictionary mapping IRI to hash of freshest terminal ontology.
        # Updated incrementally when ontologies are added.
        self._cached_ontologies: dict[str, str] = {}

    def __contains__(self, item):
        """Check if an item (IRI or ontology_id) is in the ontology manager.

        Args:
            item: The IRI or ontology_id to check.

        Returns:
            bool: True if the item exists in any version of any ontology.
        """
        # Check by IRI (primary key)
        if item in self.ontology_versions:
            return True
        # Check by ontology_id (fallback for backward compatibility)
        for versions in self.ontology_versions.values():
            for o in versions:
                if o.ontology_id == item:
                    return True
        return False

    def add_ontology(self, ontology: Ontology) -> None:
        """Add an ontology to the version tree for its IRI.

        If an ontology with the same hash already exists, it is not added again.
        The ontology is added to the version tree for its IRI.
        Ensures that created_at is set if not already present.

        Args:
            ontology: The ontology to add.
        """
        if not ontology.iri or ontology.iri == NULL_ONTOLOGY.iri:
            logger.warning(
                f"Cannot add ontology without valid IRI (ontology_id: {ontology.ontology_id})"
            )
            return

        if not ontology.hash:
            logger.warning(f"Cannot add ontology without hash (IRI: {ontology.iri})")
            return

        # Ensure created_at is set
        if not ontology.created_at:
            from datetime import datetime, timezone

            ontology.created_at = datetime.now(timezone.utc)
            logger.debug(
                f"Set created_at for ontology {ontology.iri} with hash {ontology.hash[:8]}..."
            )

        if ontology.iri not in self.ontology_versions:
            self.ontology_versions[ontology.iri] = []

        # Check if this hash already exists
        existing_hashes = {o.hash for o in self.ontology_versions[ontology.iri]}
        if ontology.hash not in existing_hashes:
            self.ontology_versions[ontology.iri].append(ontology)
            # Update cache for this specific IRI (store hash only)
            freshest = self.get_freshest_terminal_ontology_by_iri(ontology.iri)
            if freshest and freshest.hash:
                self._cached_ontologies[ontology.iri] = freshest.hash
            logger.debug(
                f"Added ontology {ontology.iri} with hash {ontology.hash[:8]}..."
            )
        else:
            logger.debug(
                f"Ontology {ontology.iri} with hash {ontology.hash[:8]}... already exists"
            )

    def get_terminal_ontologies_by_iri(self, iri: str | None = None) -> list[Ontology]:
        """Get terminal (leaf) ontologies in the version graph.

        Terminal ontologies are those that are not parents of any other ontology
        in the version tree. If iri is provided, returns terminals for
        that ontology only; otherwise returns terminals for all ontologies.

        Args:
            iri: Optional IRI to filter by.

        Returns:
            list[Ontology]: List of terminal ontologies.
        """
        if iri:
            if iri not in self.ontology_versions:
                return []
            ontologies = self.ontology_versions[iri]
        else:
            ontologies = [
                o for versions in self.ontology_versions.values() for o in versions
            ]

        if not ontologies:
            return []

        # Build a set of all parent hashes
        all_parent_hashes = set()
        for o in ontologies:
            all_parent_hashes.update(o.parent_hashes)

        # Terminal nodes are those whose hash is not in any parent_hashes
        terminal_hashes = {o.hash for o in ontologies} - all_parent_hashes

        return [o for o in ontologies if o.hash in terminal_hashes]

    def get_terminal_ontologies(self, ontology_id: str | None = None) -> list[Ontology]:
        """Get terminal (leaf) ontologies by ontology_id (backward compatibility wrapper).

        Args:
            ontology_id: Optional ontology_id to filter by.

        Returns:
            list[Ontology]: List of terminal ontologies.
        """
        if ontology_id:
            # Find IRI(s) matching this ontology_id
            matching_iris = [
                iri
                for iri, versions in self.ontology_versions.items()
                if any(o.ontology_id == ontology_id for o in versions)
            ]
            if not matching_iris:
                return []
            # Get terminals for all matching IRIs
            all_terminals = []
            for iri in matching_iris:
                all_terminals.extend(self.get_terminal_ontologies_by_iri(iri))
            return all_terminals
        else:
            return self.get_terminal_ontologies_by_iri(None)

    def get_freshest_terminal_ontology_by_iri(
        self, iri: str | None = None
    ) -> Ontology | None:
        """Get the freshest terminal ontology based on created_at timestamp.

        Returns the terminal ontology with the most recent `created_at` timestamp.
        If multiple terminal ontologies exist, returns the one that was most recently
        created. If no created_at is set, falls back to the first terminal ontology.

        Args:
            iri: Optional IRI to filter by. If None, searches across
                all ontologies.

        Returns:
            Ontology: The freshest terminal ontology, or None if no terminal
                ontologies exist.
        """
        terminals = self.get_terminal_ontologies_by_iri(iri)

        if not terminals:
            return None

        # Filter out ontologies without created_at and sort by created_at
        with_timestamp = [o for o in terminals if o.created_at is not None]
        without_timestamp = [o for o in terminals if o.created_at is None]

        if with_timestamp:
            # Sort by created_at descending (most recent first)
            # Type assertion: we know created_at is not None due to filter above
            from datetime import datetime
            from typing import cast

            freshest = max(
                with_timestamp,
                key=lambda o: cast(datetime, o.created_at),
            )
            return freshest
        elif without_timestamp:
            # Fallback to first terminal if no timestamps available
            return without_timestamp[0]

        return None

    def get_freshest_terminal_ontology(
        self, ontology_id: str | None = None
    ) -> Ontology | None:
        """Get the freshest terminal ontology by ontology_id (backward compatibility wrapper).

        Args:
            ontology_id: Optional ontology_id to filter by.

        Returns:
            Ontology: The freshest terminal ontology, or None if no terminal
                ontologies exist.
        """
        if ontology_id:
            # Find IRI(s) matching this ontology_id
            matching_iris = [
                iri
                for iri, versions in self.ontology_versions.items()
                if any(o.ontology_id == ontology_id for o in versions)
            ]
            if not matching_iris:
                return None
            # Get freshest for all matching IRIs and return the most recent
            candidates = []
            for iri in matching_iris:
                freshest = self.get_freshest_terminal_ontology_by_iri(iri)
                if freshest:
                    candidates.append(freshest)
            if not candidates:
                return None
            # Return the most recent among all candidates
            from datetime import datetime
            from typing import cast

            with_timestamp = [o for o in candidates if o.created_at is not None]
            if with_timestamp:
                return max(with_timestamp, key=lambda o: cast(datetime, o.created_at))
            return candidates[0]
        else:
            return self.get_freshest_terminal_ontology_by_iri(None)

    def get_ontology_versions_by_iri(self, iri: str) -> list[Ontology]:
        """Get all versions of an ontology by IRI.

        Args:
            iri: The IRI to retrieve versions for.

        Returns:
            list[Ontology]: List of all versions of the ontology.
        """
        return self.ontology_versions.get(iri, [])

    def get_ontology_versions(self, ontology_id: str) -> list[Ontology]:
        """Get all versions of an ontology by ontology_id (backward compatibility wrapper).

        Args:
            ontology_id: The ontology_id to retrieve versions for.

        Returns:
            list[Ontology]: List of all versions of the ontology.
        """
        # Find all IRIs matching this ontology_id
        all_versions = []
        for iri, versions in self.ontology_versions.items():
            if any(o.ontology_id == ontology_id for o in versions):
                all_versions.extend(versions)
        return all_versions

    def get_lineage_graph_by_iri(self, iri: str):
        """Get the lineage graph for a specific IRI.

        Args:
            iri: The IRI to get the lineage graph for.

        Returns:
            networkx.DiGraph: The lineage graph for the ontology, or None if not found.
        """
        if iri not in self.ontology_versions:
            return None

        return Ontology.build_lineage_graph(self.ontology_versions[iri])

    def get_lineage_graph(self, ontology_id: str):
        """Get the lineage graph for a specific ontology_id (backward compatibility wrapper).

        Args:
            ontology_id: The ontology_id to get the lineage graph for.

        Returns:
            networkx.DiGraph: The lineage graph for the ontology, or None if not found.
        """
        # Find first IRI matching this ontology_id
        for iri, versions in self.ontology_versions.items():
            if any(o.ontology_id == ontology_id for o in versions):
                return Ontology.build_lineage_graph(versions)
        return None

    def get_ontology(
        self,
        ontology_id: str | None = None,
        ontology_iri: str | None = None,
        hash: str | None = None,
    ) -> Ontology:
        """Get an ontology by its IRI, ontology_id, or hash.

        If hash is provided, returns the specific version. Otherwise, returns
        a terminal (most recent) version if multiple versions exist.
        IRI is preferred over ontology_id for lookup.

        Args:
            ontology_id: The short name of the ontology to retrieve (optional, for backward compatibility).
            ontology_iri: The IRI of the ontology to retrieve (preferred).
            hash: The hash of a specific version to retrieve (optional).

        Returns:
            Ontology: The matching ontology if found, NULL_ONTOLOGY otherwise.
        """
        # If hash is provided, search by hash first
        if hash:
            for versions in self.ontology_versions.values():
                for o in versions:
                    if o.hash == hash:
                        return o

        # Try by IRI first (preferred method)
        if ontology_iri is not None:
            if ontology_iri in self.ontology_versions:
                versions = self.ontology_versions[ontology_iri]
                if hash:
                    # Find specific version by hash
                    for o in versions:
                        if o.hash == hash:
                            return o
                else:
                    # Return terminal version (most recent)
                    terminals = self.get_terminal_ontologies_by_iri(ontology_iri)
                    if terminals:
                        return terminals[0]
                    # Fallback to first version if no terminals
                    if versions:
                        return versions[0]

        # Try by ontology_id if provided (backward compatibility)
        if ontology_id is not None:
            # Find IRI(s) matching this ontology_id
            matching_iris = [
                iri
                for iri, versions in self.ontology_versions.items()
                if any(o.ontology_id == ontology_id for o in versions)
            ]
            if matching_iris:
                # Use first matching IRI
                iri = matching_iris[0]
                versions = self.ontology_versions[iri]
                if hash:
                    # Find specific version by hash
                    for o in versions:
                        if o.hash == hash:
                            return o
                else:
                    # Return terminal version (most recent)
                    terminals = self.get_terminal_ontologies_by_iri(iri)
                    if terminals:
                        return terminals[0]
                    # Fallback to first version if no terminals
                    if versions:
                        return versions[0]

                # If IRI is also provided, check consistency
                if ontology_iri and ontology_iri != iri:
                    logger.warning(
                        f"Ontology id '{ontology_id}' matches IRI '{iri}' but different IRI '{ontology_iri}' was provided"
                    )

        # Not found
        return NULL_ONTOLOGY

    def get_ontology_iris(self) -> list[str]:
        """Get a list of all ontology IRIs.

        Returns:
            list[str]: List of ontology IRIs.
        """
        return list(self.ontology_versions.keys())

    def get_ontology_names(self) -> list[str]:
        """Get a list of all ontology short names (backward compatibility wrapper).

        Returns:
            list[str]: List of unique ontology short names.
        """
        names = set()
        for versions in self.ontology_versions.values():
            for o in versions:
                if o.ontology_id:
                    names.add(o.ontology_id)
        return sorted(list(names))

    @property
    def has_ontologies(self) -> bool:
        """Check if there are any ontologies available.

        Returns:
            bool: True if there are any ontologies, False otherwise.
        """
        return len(self._cached_ontologies) > 0 or len(self.ontology_versions) > 0

    @property
    def ontologies(self) -> list[Ontology]:
        """Get freshest terminal ontology for each IRI.

        This property provides backward compatibility with code that expects
        a list of ontologies. Returns the freshest (most recently created)
        terminal version for each IRI.

        The result is cached per IRI (as hashes) and updated incrementally
        when ontologies are added.

        Returns:
            list[Ontology]: List of freshest terminal ontologies, one per IRI.
        """
        result = []

        # Ensure cache is up to date for all IRIs
        for iri in self.ontology_versions.keys():
            if iri not in self._cached_ontologies:
                freshest = self.get_freshest_terminal_ontology_by_iri(iri)
                if freshest and freshest.hash:
                    self._cached_ontologies[iri] = freshest.hash

        # Remove entries for IRIs that no longer exist
        cached_iris = set(self._cached_ontologies.keys())
        current_iris = set(self.ontology_versions.keys())
        for removed_iri in cached_iris - current_iris:
            del self._cached_ontologies[removed_iri]

        # Look up actual ontology objects by hash
        for iri, cached_hash in self._cached_ontologies.items():
            if iri in self.ontology_versions:
                # Find ontology with matching hash
                for ontology in self.ontology_versions[iri]:
                    if ontology.hash == cached_hash:
                        result.append(ontology)
                        break

        return result

    def update_ontology(self, ontology_id: str, ontology_addendum: RDFGraph):
        """Update an existing ontology with additional triples.

        Note: This method is deprecated. Use add_ontology() with a new version
        that has the current hash in parent_hashes instead.

        Args:
            ontology_id: The short name of the ontology to update.
            ontology_addendum: The RDF graph containing additional triples to add.
        """
        logger.warning(
            "update_ontology() is deprecated. Use add_ontology() with version tracking instead."
        )
        terminals = self.get_terminal_ontologies(ontology_id)
        if terminals:
            terminals[0] += ontology_addendum
            # Update cache for the IRI (though this method is deprecated)
            iri = terminals[0].iri
            freshest = self.get_freshest_terminal_ontology_by_iri(iri)
            if freshest and freshest.hash:
                self._cached_ontologies[iri] = freshest.hash

has_ontologies property

Check if there are any ontologies available.

Returns:

Name Type Description
bool bool

True if there are any ontologies, False otherwise.

ontologies property

Get freshest terminal ontology for each IRI.

This property provides backward compatibility with code that expects a list of ontologies. Returns the freshest (most recently created) terminal version for each IRI.

The result is cached per IRI (as hashes) and updated incrementally when ontologies are added.

Returns:

Type Description
list[Ontology]

list[Ontology]: List of freshest terminal ontologies, one per IRI.

__contains__(item)

Check if an item (IRI or ontology_id) is in the ontology manager.

Parameters:

Name Type Description Default
item

The IRI or ontology_id to check.

 required

Returns:

Name Type Description
bool

True if the item exists in any version of any ontology.
Source code in ontocast/tool/ontology_manager.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __contains__(self, item):
    """Check if an item (IRI or ontology_id) is in the ontology manager.

    Args:
        item: The IRI or ontology_id to check.

    Returns:
        bool: True if the item exists in any version of any ontology.
    """
    # Check by IRI (primary key)
    if item in self.ontology_versions:
        return True
    # Check by ontology_id (fallback for backward compatibility)
    for versions in self.ontology_versions.values():
        for o in versions:
            if o.ontology_id == item:
                return True
    return False

__init__(**kwargs)

Initialize the ontology manager.

Parameters:

Name Type Description Default
**kwargs

Additional keyword arguments passed to the parent class.

 {}
Source code in ontocast/tool/ontology_manager.py 
35
36
37
38
39
40
41
42
43
44
def __init__(self, **kwargs):
    """Initialize the ontology manager.

    Args:
        **kwargs: Additional keyword arguments passed to the parent class.
    """
    super().__init__(**kwargs)
    # Cache dictionary mapping IRI to hash of freshest terminal ontology.
    # Updated incrementally when ontologies are added.
    self._cached_ontologies: dict[str, str] = {}

add_ontology(ontology)

Add an ontology to the version tree for its IRI.

If an ontology with the same hash already exists, it is not added again. The ontology is added to the version tree for its IRI. Ensures that created_at is set if not already present.

Parameters:

Name Type Description Default
ontology Ontology

The ontology to add.

 required
Source code in ontocast/tool/ontology_manager.py 
 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
def add_ontology(self, ontology: Ontology) -> None:
    """Add an ontology to the version tree for its IRI.

    If an ontology with the same hash already exists, it is not added again.
    The ontology is added to the version tree for its IRI.
    Ensures that created_at is set if not already present.

    Args:
        ontology: The ontology to add.
    """
    if not ontology.iri or ontology.iri == NULL_ONTOLOGY.iri:
        logger.warning(
            f"Cannot add ontology without valid IRI (ontology_id: {ontology.ontology_id})"
        )
        return

    if not ontology.hash:
        logger.warning(f"Cannot add ontology without hash (IRI: {ontology.iri})")
        return

    # Ensure created_at is set
    if not ontology.created_at:
        from datetime import datetime, timezone

        ontology.created_at = datetime.now(timezone.utc)
        logger.debug(
            f"Set created_at for ontology {ontology.iri} with hash {ontology.hash[:8]}..."
        )

    if ontology.iri not in self.ontology_versions:
        self.ontology_versions[ontology.iri] = []

    # Check if this hash already exists
    existing_hashes = {o.hash for o in self.ontology_versions[ontology.iri]}
    if ontology.hash not in existing_hashes:
        self.ontology_versions[ontology.iri].append(ontology)
        # Update cache for this specific IRI (store hash only)
        freshest = self.get_freshest_terminal_ontology_by_iri(ontology.iri)
        if freshest and freshest.hash:
            self._cached_ontologies[ontology.iri] = freshest.hash
        logger.debug(
            f"Added ontology {ontology.iri} with hash {ontology.hash[:8]}..."
        )
    else:
        logger.debug(
            f"Ontology {ontology.iri} with hash {ontology.hash[:8]}... already exists"
        )

get_freshest_terminal_ontology(ontology_id=None)

Get the freshest terminal ontology by ontology_id (backward compatibility wrapper).

Parameters:

Name Type Description Default
ontology_id str | None

Optional ontology_id to filter by.

 None

Returns:

Name Type Description
Ontology Ontology | None

The freshest terminal ontology, or None if no terminal ontologies exist.
Source code in ontocast/tool/ontology_manager.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def get_freshest_terminal_ontology(
    self, ontology_id: str | None = None
) -> Ontology | None:
    """Get the freshest terminal ontology by ontology_id (backward compatibility wrapper).

    Args:
        ontology_id: Optional ontology_id to filter by.

    Returns:
        Ontology: The freshest terminal ontology, or None if no terminal
            ontologies exist.
    """
    if ontology_id:
        # Find IRI(s) matching this ontology_id
        matching_iris = [
            iri
            for iri, versions in self.ontology_versions.items()
            if any(o.ontology_id == ontology_id for o in versions)
        ]
        if not matching_iris:
            return None
        # Get freshest for all matching IRIs and return the most recent
        candidates = []
        for iri in matching_iris:
            freshest = self.get_freshest_terminal_ontology_by_iri(iri)
            if freshest:
                candidates.append(freshest)
        if not candidates:
            return None
        # Return the most recent among all candidates
        from datetime import datetime
        from typing import cast

        with_timestamp = [o for o in candidates if o.created_at is not None]
        if with_timestamp:
            return max(with_timestamp, key=lambda o: cast(datetime, o.created_at))
        return candidates[0]
    else:
        return self.get_freshest_terminal_ontology_by_iri(None)

get_freshest_terminal_ontology_by_iri(iri=None)

Get the freshest terminal ontology based on created_at timestamp.

Returns the terminal ontology with the most recent created_at timestamp. If multiple terminal ontologies exist, returns the one that was most recently created. If no created_at is set, falls back to the first terminal ontology.

Parameters:

Name Type Description Default
iri str | None

Optional IRI to filter by. If None, searches across all ontologies.

 None

Returns:

Name Type Description
Ontology Ontology | None

The freshest terminal ontology, or None if no terminal ontologies exist.
Source code in ontocast/tool/ontology_manager.py 
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
203
204
205
206
207
208
209
210
211
212
213
214
215
def get_freshest_terminal_ontology_by_iri(
    self, iri: str | None = None
) -> Ontology | None:
    """Get the freshest terminal ontology based on created_at timestamp.

    Returns the terminal ontology with the most recent `created_at` timestamp.
    If multiple terminal ontologies exist, returns the one that was most recently
    created. If no created_at is set, falls back to the first terminal ontology.

    Args:
        iri: Optional IRI to filter by. If None, searches across
            all ontologies.

    Returns:
        Ontology: The freshest terminal ontology, or None if no terminal
            ontologies exist.
    """
    terminals = self.get_terminal_ontologies_by_iri(iri)

    if not terminals:
        return None

    # Filter out ontologies without created_at and sort by created_at
    with_timestamp = [o for o in terminals if o.created_at is not None]
    without_timestamp = [o for o in terminals if o.created_at is None]

    if with_timestamp:
        # Sort by created_at descending (most recent first)
        # Type assertion: we know created_at is not None due to filter above
        from datetime import datetime
        from typing import cast

        freshest = max(
            with_timestamp,
            key=lambda o: cast(datetime, o.created_at),
        )
        return freshest
    elif without_timestamp:
        # Fallback to first terminal if no timestamps available
        return without_timestamp[0]

    return None

get_lineage_graph(ontology_id)

Get the lineage graph for a specific ontology_id (backward compatibility wrapper).

Parameters:

Name Type Description Default
ontology_id str

The ontology_id to get the lineage graph for.

 required

Returns:

Type Description

networkx.DiGraph: The lineage graph for the ontology, or None if not found.
Source code in ontocast/tool/ontology_manager.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def get_lineage_graph(self, ontology_id: str):
    """Get the lineage graph for a specific ontology_id (backward compatibility wrapper).

    Args:
        ontology_id: The ontology_id to get the lineage graph for.

    Returns:
        networkx.DiGraph: The lineage graph for the ontology, or None if not found.
    """
    # Find first IRI matching this ontology_id
    for iri, versions in self.ontology_versions.items():
        if any(o.ontology_id == ontology_id for o in versions):
            return Ontology.build_lineage_graph(versions)
    return None

get_lineage_graph_by_iri(iri)

Get the lineage graph for a specific IRI.

Parameters:

Name Type Description Default
iri str

The IRI to get the lineage graph for.

 required

Returns:

Type Description

networkx.DiGraph: The lineage graph for the ontology, or None if not found.
Source code in ontocast/tool/ontology_manager.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
def get_lineage_graph_by_iri(self, iri: str):
    """Get the lineage graph for a specific IRI.

    Args:
        iri: The IRI to get the lineage graph for.

    Returns:
        networkx.DiGraph: The lineage graph for the ontology, or None if not found.
    """
    if iri not in self.ontology_versions:
        return None

    return Ontology.build_lineage_graph(self.ontology_versions[iri])

get_ontology(ontology_id=None, ontology_iri=None, hash=None)

Get an ontology by its IRI, ontology_id, or hash.

If hash is provided, returns the specific version. Otherwise, returns a terminal (most recent) version if multiple versions exist. IRI is preferred over ontology_id for lookup.

Parameters:

Name Type Description Default
ontology_id str | None

The short name of the ontology to retrieve (optional, for backward compatibility).

 None
ontology_iri str | None

The IRI of the ontology to retrieve (preferred).

 None
hash str | None

The hash of a specific version to retrieve (optional).

 None

Returns:

Name Type Description
Ontology Ontology

The matching ontology if found, NULL_ONTOLOGY otherwise.
Source code in ontocast/tool/ontology_manager.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
def get_ontology(
    self,
    ontology_id: str | None = None,
    ontology_iri: str | None = None,
    hash: str | None = None,
) -> Ontology:
    """Get an ontology by its IRI, ontology_id, or hash.

    If hash is provided, returns the specific version. Otherwise, returns
    a terminal (most recent) version if multiple versions exist.
    IRI is preferred over ontology_id for lookup.

    Args:
        ontology_id: The short name of the ontology to retrieve (optional, for backward compatibility).
        ontology_iri: The IRI of the ontology to retrieve (preferred).
        hash: The hash of a specific version to retrieve (optional).

    Returns:
        Ontology: The matching ontology if found, NULL_ONTOLOGY otherwise.
    """
    # If hash is provided, search by hash first
    if hash:
        for versions in self.ontology_versions.values():
            for o in versions:
                if o.hash == hash:
                    return o

    # Try by IRI first (preferred method)
    if ontology_iri is not None:
        if ontology_iri in self.ontology_versions:
            versions = self.ontology_versions[ontology_iri]
            if hash:
                # Find specific version by hash
                for o in versions:
                    if o.hash == hash:
                        return o
            else:
                # Return terminal version (most recent)
                terminals = self.get_terminal_ontologies_by_iri(ontology_iri)
                if terminals:
                    return terminals[0]
                # Fallback to first version if no terminals
                if versions:
                    return versions[0]

    # Try by ontology_id if provided (backward compatibility)
    if ontology_id is not None:
        # Find IRI(s) matching this ontology_id
        matching_iris = [
            iri
            for iri, versions in self.ontology_versions.items()
            if any(o.ontology_id == ontology_id for o in versions)
        ]
        if matching_iris:
            # Use first matching IRI
            iri = matching_iris[0]
            versions = self.ontology_versions[iri]
            if hash:
                # Find specific version by hash
                for o in versions:
                    if o.hash == hash:
                        return o
            else:
                # Return terminal version (most recent)
                terminals = self.get_terminal_ontologies_by_iri(iri)
                if terminals:
                    return terminals[0]
                # Fallback to first version if no terminals
                if versions:
                    return versions[0]

            # If IRI is also provided, check consistency
            if ontology_iri and ontology_iri != iri:
                logger.warning(
                    f"Ontology id '{ontology_id}' matches IRI '{iri}' but different IRI '{ontology_iri}' was provided"
                )

    # Not found
    return NULL_ONTOLOGY

get_ontology_iris()

Get a list of all ontology IRIs.

Returns:

Type Description
list[str]

list[str]: List of ontology IRIs.
Source code in ontocast/tool/ontology_manager.py 
393
394
395
396
397
398
399
def get_ontology_iris(self) -> list[str]:
    """Get a list of all ontology IRIs.

    Returns:
        list[str]: List of ontology IRIs.
    """
    return list(self.ontology_versions.keys())

get_ontology_names()

Get a list of all ontology short names (backward compatibility wrapper).

Returns:

Type Description
list[str]

list[str]: List of unique ontology short names.
Source code in ontocast/tool/ontology_manager.py 
401
402
403
404
405
406
407
408
409
410
411
412
def get_ontology_names(self) -> list[str]:
    """Get a list of all ontology short names (backward compatibility wrapper).

    Returns:
        list[str]: List of unique ontology short names.
    """
    names = set()
    for versions in self.ontology_versions.values():
        for o in versions:
            if o.ontology_id:
                names.add(o.ontology_id)
    return sorted(list(names))

get_ontology_versions(ontology_id)

Get all versions of an ontology by ontology_id (backward compatibility wrapper).

Parameters:

Name Type Description Default
ontology_id str

The ontology_id to retrieve versions for.

 required

Returns:

Type Description
list[Ontology]

list[Ontology]: List of all versions of the ontology.
Source code in ontocast/tool/ontology_manager.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def get_ontology_versions(self, ontology_id: str) -> list[Ontology]:
    """Get all versions of an ontology by ontology_id (backward compatibility wrapper).

    Args:
        ontology_id: The ontology_id to retrieve versions for.

    Returns:
        list[Ontology]: List of all versions of the ontology.
    """
    # Find all IRIs matching this ontology_id
    all_versions = []
    for iri, versions in self.ontology_versions.items():
        if any(o.ontology_id == ontology_id for o in versions):
            all_versions.extend(versions)
    return all_versions

get_ontology_versions_by_iri(iri)

Get all versions of an ontology by IRI.

Parameters:

Name Type Description Default
iri str

The IRI to retrieve versions for.

 required

Returns:

Type Description
list[Ontology]

list[Ontology]: List of all versions of the ontology.
Source code in ontocast/tool/ontology_manager.py 
257
258
259
260
261
262
263
264
265
266
def get_ontology_versions_by_iri(self, iri: str) -> list[Ontology]:
    """Get all versions of an ontology by IRI.

    Args:
        iri: The IRI to retrieve versions for.

    Returns:
        list[Ontology]: List of all versions of the ontology.
    """
    return self.ontology_versions.get(iri, [])

get_terminal_ontologies(ontology_id=None)

Get terminal (leaf) ontologies by ontology_id (backward compatibility wrapper).

Parameters:

Name Type Description Default
ontology_id str | None

Optional ontology_id to filter by.

 None

Returns:

Type Description
list[Ontology]

list[Ontology]: List of terminal ontologies.
Source code in ontocast/tool/ontology_manager.py 
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
def get_terminal_ontologies(self, ontology_id: str | None = None) -> list[Ontology]:
    """Get terminal (leaf) ontologies by ontology_id (backward compatibility wrapper).

    Args:
        ontology_id: Optional ontology_id to filter by.

    Returns:
        list[Ontology]: List of terminal ontologies.
    """
    if ontology_id:
        # Find IRI(s) matching this ontology_id
        matching_iris = [
            iri
            for iri, versions in self.ontology_versions.items()
            if any(o.ontology_id == ontology_id for o in versions)
        ]
        if not matching_iris:
            return []
        # Get terminals for all matching IRIs
        all_terminals = []
        for iri in matching_iris:
            all_terminals.extend(self.get_terminal_ontologies_by_iri(iri))
        return all_terminals
    else:
        return self.get_terminal_ontologies_by_iri(None)

get_terminal_ontologies_by_iri(iri=None)

Get terminal (leaf) ontologies in the version graph.

Terminal ontologies are those that are not parents of any other ontology in the version tree. If iri is provided, returns terminals for that ontology only; otherwise returns terminals for all ontologies.

Parameters:

Name Type Description Default
iri str | None

Optional IRI to filter by.

 None

Returns:

Type Description
list[Ontology]

list[Ontology]: List of terminal ontologies.
Source code in ontocast/tool/ontology_manager.py 
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
def get_terminal_ontologies_by_iri(self, iri: str | None = None) -> list[Ontology]:
    """Get terminal (leaf) ontologies in the version graph.

    Terminal ontologies are those that are not parents of any other ontology
    in the version tree. If iri is provided, returns terminals for
    that ontology only; otherwise returns terminals for all ontologies.

    Args:
        iri: Optional IRI to filter by.

    Returns:
        list[Ontology]: List of terminal ontologies.
    """
    if iri:
        if iri not in self.ontology_versions:
            return []
        ontologies = self.ontology_versions[iri]
    else:
        ontologies = [
            o for versions in self.ontology_versions.values() for o in versions
        ]

    if not ontologies:
        return []

    # Build a set of all parent hashes
    all_parent_hashes = set()
    for o in ontologies:
        all_parent_hashes.update(o.parent_hashes)

    # Terminal nodes are those whose hash is not in any parent_hashes
    terminal_hashes = {o.hash for o in ontologies} - all_parent_hashes

    return [o for o in ontologies if o.hash in terminal_hashes]

update_ontology(ontology_id, ontology_addendum)

Update an existing ontology with additional triples.

Note: This method is deprecated. Use add_ontology() with a new version that has the current hash in parent_hashes instead.

Parameters:

Name Type Description Default
ontology_id str

The short name of the ontology to update.

 required
ontology_addendum RDFGraph

The RDF graph containing additional triples to add.

 required
Source code in ontocast/tool/ontology_manager.py 
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
def update_ontology(self, ontology_id: str, ontology_addendum: RDFGraph):
    """Update an existing ontology with additional triples.

    Note: This method is deprecated. Use add_ontology() with a new version
    that has the current hash in parent_hashes instead.

    Args:
        ontology_id: The short name of the ontology to update.
        ontology_addendum: The RDF graph containing additional triples to add.
    """
    logger.warning(
        "update_ontology() is deprecated. Use add_ontology() with version tracking instead."
    )
    terminals = self.get_terminal_ontologies(ontology_id)
    if terminals:
        terminals[0] += ontology_addendum
        # Update cache for the IRI (though this method is deprecated)
        iri = terminals[0].iri
        freshest = self.get_freshest_terminal_ontology_by_iri(iri)
        if freshest and freshest.hash:
            self._cached_ontologies[iri] = freshest.hash

Tool

Bases: BasePydanticModel

Base class for all OntoCast tools.

This class serves as the foundation for all tools in the OntoCast system. It provides common functionality and interface that all tools must implement. Tools should inherit from this class and implement their specific functionality.

Source code in ontocast/tool/onto.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class Tool(BasePydanticModel):
    """Base class for all OntoCast tools.

    This class serves as the foundation for all tools in the OntoCast system.
    It provides common functionality and interface that all tools must implement.
    Tools should inherit from this class and implement their specific functionality.

    Attributes:
        Inherits all attributes from BasePydanticModel.
    """

    def __init__(self, **kwargs):
        """Initialize the tool.

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

__init__(**kwargs)

Initialize the tool.

Parameters:

Name Type Description Default
**kwargs

Keyword arguments passed to the parent class.

 {}
Source code in ontocast/tool/onto.py 
25
26
27
28
29
30
31
def __init__(self, **kwargs):
    """Initialize the tool.

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

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