From b6d6e8e4f5d35a3a068e091abdf4f213894b31c5 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 11:09:22 -0400 Subject: [PATCH 01/20] new: reset_arango_db --- tests/conftest.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index 6a51a160..506821c3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import os +import pytest import subprocess from pathlib import Path from typing import Any, Dict, Set, Tuple @@ -53,6 +54,17 @@ class NoTimeoutHTTPClient(DefaultHTTPClient): adbrdf = ArangoRDF(db) +@pytest.fixture(autouse=True) +def reset_arango_db(): + global db + for g in db.graphs(): + db.delete_graph(g["name"], drop_collections=True) + + for c in db.collections(): + if c["system"] == False: + db.delete_collection(c["name"]) + + def arango_restore(path_to_data: str) -> None: global con restore_prefix = "./tools/" if os.getenv("GITHUB_ACTIONS") else "" From 663096ea674aed8a210916c24b21b595bda7f669 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 11:09:35 -0400 Subject: [PATCH 02/20] new: skip write_adb_col if `resource_collection` is defined --- arango_rdf/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 270d317d..06984a09 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -1014,7 +1014,7 @@ def rdf_to_arangodb_by_pgt( if overwrite_graph: self.db.delete_graph(name, ignore_missing=True, drop_collections=True) - if write_adb_col_statements or contextualize_graph: + if self.__resource_collection is None and (write_adb_col_statements or contextualize_graph): # Enabling Graph Contextualization forces # us to run the ArangoDB Collection Mapping algorithm # regardless of **write_adb_col_statements** From b74776ef28b67509c3f61a5c605084eeb4256105 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 11:23:35 -0400 Subject: [PATCH 03/20] v2: `__pgt_parse_literal_statements` --- arango_rdf/main.py | 57 +++++++++++++++++++++++++++------------------- 1 file changed, 33 insertions(+), 24 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 06984a09..f6f70804 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -1014,7 +1014,8 @@ def rdf_to_arangodb_by_pgt( if overwrite_graph: self.db.delete_graph(name, ignore_missing=True, drop_collections=True) - if self.__resource_collection is None and (write_adb_col_statements or contextualize_graph): + # if self.__resource_collection is None: + if write_adb_col_statements or contextualize_graph: # Enabling Graph Contextualization forces # us to run the ArangoDB Collection Mapping algorithm # regardless of **write_adb_col_statements** @@ -2349,29 +2350,37 @@ def __pgt_parse_literal_statements( `ArangoRDF.__insert_adb_docs()`. :type adb_import_kwargs: Dict[str, Any] """ - # TODO: Revisit FILTER clauses - # We rely on the FILTER clauses to make sure no literal - # statements belonging to RDF Lists are processed, - # as that is handled in another step (which also needs work...) - query = f""" - PREFIX rdf: <{RDF}> - - SELECT ?subject ?predicate - WHERE {{ - ?subject ?predicate ?object . - FILTER isLiteral(?object) - FILTER NOT EXISTS {{ ?subject rdf:first ?anyObject }} - FILTER NOT EXISTS {{ ?subject rdf:rest ?anyObject }} - FILTER (!regex(str(?predicate), "^{RDF}_[0-9]+$")) - FILTER (!regex(str(?predicate), "^{RDF}li$")) - }} - GROUP BY ?subject ?predicate - """ - - with get_spinner_progress("(RDF → ADB): PGT [RDF Literals (Query)]") as sp: - sp.add_task("") + # Pre-compute blacklist sets + rdf_list_subjects = set() + + # Collect RDF list subjects + for s in self.__rdf_graph.subjects(RDF.first, None): + rdf_list_subjects.add(s) + for s in self.__rdf_graph.subjects(RDF.rest, None): + rdf_list_subjects.add(s) + + # Use string patterns (same as your __pgt_statement_is_part_of_rdf_list method) + rdf_ns = str(RDF) + container_pattern_n = re.compile(f"^{re.escape(rdf_ns)}_[0-9]+$") + container_pattern_li = re.compile(f"^{re.escape(rdf_ns)}li$") + + # Direct iteration with efficient filtering + literal_pairs = set() # Use set to avoid duplicates automatically + + for s, p, o in self.__rdf_graph: + if not isinstance(o, Literal): + continue - data = self.__rdf_graph.query(query) + if s in rdf_list_subjects: + continue + + p_str = str(p) + if container_pattern_n.match(p_str) or container_pattern_li.match(p_str): + continue + + literal_pairs.add((s, p)) + + data = list(literal_pairs) s: RDFTerm p: URIRef @@ -2380,7 +2389,7 @@ def __pgt_parse_literal_statements( total = len(data) batch_size = batch_size or total bar_progress = get_bar_progress("(RDF → ADB): PGT [RDF Literals]", "#EF7D00") - bar_progress_task = bar_progress.add_task("", total=total - 1) + bar_progress_task = bar_progress.add_task("", total=total) spinner_progress = get_import_spinner_progress(" ") statements = ( From 3e51b735b770f5bb726161a1c9d69a9aa4401784 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 16:12:46 -0400 Subject: [PATCH 04/20] new: early exit on `write_adb_cols` --- arango_rdf/main.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index f6f70804..9a1263de 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -1014,7 +1014,6 @@ def rdf_to_arangodb_by_pgt( if overwrite_graph: self.db.delete_graph(name, ignore_missing=True, drop_collections=True) - # if self.__resource_collection is None: if write_adb_col_statements or contextualize_graph: # Enabling Graph Contextualization forces # us to run the ArangoDB Collection Mapping algorithm @@ -1126,6 +1125,7 @@ def write_adb_col_statements( rdf_graph: RDFGraph, adb_col_statements: Optional[RDFGraph] = None, uri_map_collection_name: Optional[str] = None, + ) -> RDFGraph: """RDF -> ArangoDB (PGT): Run the ArangoDB Collection Mapping Process for **rdf_graph** to map RDF Resources to their respective ArangoDB Collection. @@ -1198,6 +1198,13 @@ def write_adb_col_statements( if self.__contextualize_graph: self.__type_map = self.__combine_type_map_and_dr_map() + # If the resource collection is not None, we don't need to run the + # ArangoDB Collection Mapping Process to completion, since we will + # be using the resource collection for all RDF Resources except for + # Class and Property. + if self.__resource_collection is not None: + return self.__adb_col_statements + # 5. Finalize **adb_col_statements** for rdf_map in [self.__explicit_type_map, self.__domain_range_map]: for rdf_resource, class_set in rdf_map.items(): From ff8c7f48fea9fce781f99b0e6a1bba3d86f30cd1 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 17:19:28 -0400 Subject: [PATCH 05/20] new: `term_metadata_cache`, `insert_async`, cleanup --- arango_rdf/main.py | 130 ++++++++++++++++++++++++++------------------- 1 file changed, 76 insertions(+), 54 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 9a1263de..8d2a8c98 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -12,7 +12,7 @@ import farmhash from arango.collection import StandardCollection from arango.cursor import Cursor -from arango.database import StandardDatabase +from arango.database import AsyncDatabase, StandardDatabase from arango.graph import Graph as ADBGraph from isodate import Duration from rdflib import RDF, RDFS, XSD, BNode @@ -77,6 +77,7 @@ def __init__( controller: ArangoRDFController = ArangoRDFController(), logging_lvl: Union[str, int] = logging.INFO, rdf_attribute_prefix: str = "_", + insert_async: bool = False, ): self.set_logging(logging_lvl) @@ -88,9 +89,13 @@ def __init__( msg = "**controller** parameter must inherit from ArangoRDFController" raise TypeError(msg) - self.__db = db - self.__cntrl = controller - self.__cntrl.db = db + self.db: StandardDatabase = db + self.async_db: AsyncDatabase = db.begin_async_execution(return_result=False) + self.insert_async = insert_async + + self.controller: ArangoRDFController = controller + self.controller.db: StandardDatabase = db + self.controller.async_db: AsyncDatabase = self.async_db # Set the RDF attribute prefix self.__rdf_attribute_prefix = rdf_attribute_prefix @@ -141,6 +146,9 @@ def __init__( # A mapping of Reified Subjects to their corresponding ArangoDB Edge. self.__reified_subject_map: Dict[Union[URIRef, BNode], Tuple[str, str, str]] + # Metadata cache for performance optimization + self.__term_metadata_cache: Dict[str, RDFTermMeta] = {} + # Commonly used URIs self.__rdfs_resource_str = str(RDFS.Resource) self.__rdfs_class_str = str(RDFS.Class) @@ -163,14 +171,6 @@ def __init__( logger.info(f"Instantiated ArangoRDF with database '{db.name}'") - @property - def db(self) -> StandardDatabase: - return self.__db # pragma: no cover - - @property - def controller(self) -> ArangoRDFController: - return self.__cntrl # pragma: no cover - @property def rdf_attribute_prefix(self) -> str: return self.__rdf_attribute_prefix # pragma: no cover @@ -916,6 +916,8 @@ def rdf_to_arangodb_by_pgt( """ raise TypeError(m) + self.clear_term_metadata_cache() + namespace_prefixes = [] if namespace_collection_name: namespace_prefixes = [ @@ -930,26 +932,24 @@ def rdf_to_arangodb_by_pgt( m = "Cannot specify both **uri_map_collection_name** and **resource_collection_name**." # noqa: E501 raise ValueError(m) - if not self.__db.has_collection(uri_map_collection_name): - self.__db.create_collection(uri_map_collection_name) + if not self.db.has_collection(uri_map_collection_name): + self.db.create_collection(uri_map_collection_name) - self.__uri_map_collection = self.__db.collection(uri_map_collection_name) + self.__uri_map_collection = self.db.collection(uri_map_collection_name) self.__resource_collection = None if resource_collection_name: - if not self.__db.has_collection(resource_collection_name): - self.__db.create_collection(resource_collection_name) + if not self.db.has_collection(resource_collection_name): + self.db.create_collection(resource_collection_name) - self.__resource_collection = self.__db.collection(resource_collection_name) + self.__resource_collection = self.db.collection(resource_collection_name) self.__predicate_collection = None if predicate_collection_name: - if not self.__db.has_collection(predicate_collection_name): - self.__db.create_collection(predicate_collection_name, edge=True) + if not self.db.has_collection(predicate_collection_name): + self.db.create_collection(predicate_collection_name, edge=True) - self.__predicate_collection = self.__db.collection( - predicate_collection_name - ) + self.__predicate_collection = self.db.collection(predicate_collection_name) # Reset the ArangoDB Config self.__adb_docs = defaultdict(lambda: defaultdict(dict)) @@ -1097,27 +1097,29 @@ def rdf_to_arangodb_by_pgt( bar_progress = get_bar_progress("(RDF → ADB): PGT [RDF Lists]", "#EF7D00") with Live(Group(bar_progress, spinner_progress)): self.__pgt_process_rdf_lists(bar_progress) - self.__insert_adb_docs(spinner_progress) + self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) ########################### # PGT: Namespace Prefixes # ########################### if namespace_collection_name: - if not self.__db.has_collection(namespace_collection_name): - self.__db.create_collection(namespace_collection_name) + if not self.db.has_collection(namespace_collection_name): + self.db.create_collection(namespace_collection_name) docs = [ {"prefix": prefix, "uri": uri, "_key": self.hash(uri)} for prefix, uri in namespace_prefixes ] - result = self.__db.collection(namespace_collection_name).insert_many( + result = self.db.collection(namespace_collection_name).insert_many( docs, overwrite=True, raise_on_document_error=True ) logger.debug(result) + self.clear_term_metadata_cache() + return self.__pgt_create_adb_graph(name) def write_adb_col_statements( @@ -1125,7 +1127,6 @@ def write_adb_col_statements( rdf_graph: RDFGraph, adb_col_statements: Optional[RDFGraph] = None, uri_map_collection_name: Optional[str] = None, - ) -> RDFGraph: """RDF -> ArangoDB (PGT): Run the ArangoDB Collection Mapping Process for **rdf_graph** to map RDF Resources to their respective ArangoDB Collection. @@ -1161,18 +1162,18 @@ def write_adb_col_statements( self.__adb_col_statements.bind("adb", self.__adb_ns) self.__rdf_graph = rdf_graph - self.__cntrl.rdf_graph = rdf_graph + self.controller.rdf_graph = rdf_graph with get_spinner_progress("(RDF → ADB): Write Col Statements") as rp: rp.add_task("") # 0. Add URI Collection statements if uri_map_collection_name: - if not self.__db.has_collection(uri_map_collection_name): + if not self.db.has_collection(uri_map_collection_name): m = f"URI collection '{uri_map_collection_name}' does not exist" raise ValueError(m) - for doc in self.__db.collection(uri_map_collection_name): + for doc in self.db.collection(uri_map_collection_name): uri = URIRef(doc[self.__rdf_uri_attr]) collection = str(doc["collection"]) self.__add_adb_col_statement(uri, collection, True) @@ -1212,7 +1213,7 @@ def write_adb_col_statements( if t in self.__adb_col_statements or len(class_set) == 0: continue # pragma: no cover # (false negative) - best_class = self.__cntrl.identify_best_class( + best_class = self.controller.identify_best_class( rdf_resource, class_set, self.__subclass_tree ) @@ -1254,8 +1255,8 @@ def migrate_unknown_resources( """ ur_collection_name = f"{graph_name}_UnknownResource" - ur_collection = self.__db.collection(ur_collection_name) - uri_map_collection = self.__db.collection(uri_map_collection_name) + ur_collection = self.db.collection(ur_collection_name) + uri_map_collection = self.db.collection(uri_map_collection_name) if ur_collection.count() == 0: logger.info("No Unknown Resources to migrate") @@ -1300,9 +1301,7 @@ def migrate_unknown_resources( "graph": graph_name, } - cursor = self.__db.aql.execute( - query, bind_vars=bind_vars, stream=True, **kwargs - ) + cursor = self.db.aql.execute(query, bind_vars=bind_vars, stream=True, **kwargs) edge_count = 0 @@ -1319,7 +1318,7 @@ def migrate_unknown_resources( edges_to_modify = edge_data["edges_to_modify"] edge_count += len(edges_to_modify) - result = self.__db.collection(edge_collection).update_many( + result = self.db.collection(edge_collection).update_many( edges_to_modify, merge=True, raise_on_document_error=True, @@ -1327,9 +1326,7 @@ def migrate_unknown_resources( logger.debug(result) - self.__db.collection(collection).update( - data, merge=True, silent=True - ) + self.db.collection(collection).update(data, merge=True, silent=True) cursor.batch().clear() if cursor.has_more(): @@ -1640,12 +1637,12 @@ def __fetch_adb_docs( if ignored_attributes: aql_return_value = f"UNSET(doc, {list(ignored_attributes)})" - col_size: int = self.__db.collection(col).count() + col_size: int = self.db.collection(col).count() with get_spinner_progress(f"(ADB → RDF): Export '{col}' ({col_size})") as sp: sp.add_task("") - cursor: Cursor = self.__db.aql.execute( + cursor: Cursor = self.db.aql.execute( f"FOR doc IN @@col RETURN {aql_return_value}", bind_vars={"@col": col}, **{**adb_export_kwargs, **{"stream": True}}, @@ -2359,34 +2356,34 @@ def __pgt_parse_literal_statements( """ # Pre-compute blacklist sets rdf_list_subjects = set() - + # Collect RDF list subjects for s in self.__rdf_graph.subjects(RDF.first, None): rdf_list_subjects.add(s) for s in self.__rdf_graph.subjects(RDF.rest, None): rdf_list_subjects.add(s) - + # Use string patterns (same as your __pgt_statement_is_part_of_rdf_list method) rdf_ns = str(RDF) container_pattern_n = re.compile(f"^{re.escape(rdf_ns)}_[0-9]+$") container_pattern_li = re.compile(f"^{re.escape(rdf_ns)}li$") - + # Direct iteration with efficient filtering literal_pairs = set() # Use set to avoid duplicates automatically - + for s, p, o in self.__rdf_graph: if not isinstance(o, Literal): continue if s in rdf_list_subjects: continue - + p_str = str(p) if container_pattern_n.match(p_str) or container_pattern_li.match(p_str): continue - + literal_pairs.add((s, p)) - + data = list(literal_pairs) s: RDFTerm @@ -2405,15 +2402,22 @@ def __pgt_parse_literal_statements( else self.__rdf_graph.triples ) + # Optimization: track processed terms to avoid duplicate work + processed_terms = set() + with Live(Group(bar_progress, spinner_progress)): for i, (s, p) in enumerate(data, 1): bar_progress.update(bar_progress_task, advance=1) s_meta = self.__pgt_get_term_metadata(s) + # if s not in processed_terms: self.__pgt_process_rdf_term(s_meta) + # processed_terms.add(s) p_meta = self.__pgt_get_term_metadata(p) + # if p not in processed_terms: self.__pgt_process_rdf_term(p_meta) + # processed_terms.add(p) _, s_col, s_key, _ = s_meta _, _, _, p_label = p_meta @@ -2491,10 +2495,16 @@ def __pgt_get_term_metadata(self, t: Union[URIRef, BNode, Literal]) -> RDFTermMe Collection name, Document Key, and Document label. :rtype: Tuple[URIRef | BNode | Literal, str, str, str] """ + # Quick return for Literals (no caching needed) if type(t) is Literal: return t, "", "", "" # No other metadata needed + # Check cache first t_str = str(t) + if t_str in self.__term_metadata_cache: + return self.__term_metadata_cache[t_str] + + # Compute metadata if not cached t_col = "" t_key = self.rdf_id_to_adb_key(t_str, t) t_label = self.rdf_id_to_adb_label(t_str) @@ -2522,7 +2532,18 @@ def __pgt_get_term_metadata(self, t: Union[URIRef, BNode, Literal]) -> RDFTermMe logger.debug(f"Found unknown resource: {t} ({t_key})") t_col = self.__UNKNOWN_RESOURCE - return t, str(t_col), t_key, t_label + # Cache the result and return + result = t, str(t_col), t_key, t_label + self.__term_metadata_cache[t_str] = result + return result + + def clear_term_metadata_cache(self) -> None: + """Clear the metadata cache to free memory. + + Useful for long-running operations or when processing multiple + graphs to prevent excessive memory usage. + """ + self.__term_metadata_cache.clear() def __pgt_rdf_val_to_adb_val( self, @@ -3666,8 +3687,11 @@ def __insert_adb_docs( if len(self.__adb_docs) == 0: return + db = self.async_db if self.insert_async else self.db + adb_import_kwargs["overwrite_mode"] = "update" adb_import_kwargs["merge"] = True + if "raise_on_document_error" not in adb_import_kwargs: adb_import_kwargs["raise_on_document_error"] = True @@ -3687,9 +3711,7 @@ def __insert_adb_docs( logger.debug(f"Inserting Documents: {doc_list}") try: - result = self.db.collection(col).insert_many( - doc_list, **adb_import_kwargs - ) + result = db.collection(col).insert_many(doc_list, **adb_import_kwargs) except Exception as e: e_str = str(e) From 8e97eec65560765f2ad3a5125059e7a66d081e7a Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 17:19:33 -0400 Subject: [PATCH 06/20] async_db --- arango_rdf/controller.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/arango_rdf/controller.py b/arango_rdf/controller.py index 55cf7aa1..9f52f79f 100644 --- a/arango_rdf/controller.py +++ b/arango_rdf/controller.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 from typing import Set -from arango.database import StandardDatabase +from arango.database import AsyncDatabase, StandardDatabase from rdflib import Graph from .abc import AbstractArangoRDFController @@ -27,6 +27,7 @@ class ArangoRDFController(AbstractArangoRDFController): def __init__(self) -> None: self.db: StandardDatabase + self.async_db: AsyncDatabase self.rdf_graph: Graph def identify_best_class( From f6dcf2d4e070e5a7af81287c76361167d3f0f860 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 18:13:35 -0400 Subject: [PATCH 07/20] new: `__precompute_rdf_list_info`, `__is_rdf_list_statement` --- arango_rdf/main.py | 169 +++++++++++++++++++++------------------------ 1 file changed, 77 insertions(+), 92 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 8d2a8c98..01015814 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -109,6 +109,11 @@ def __init__( self.__rdf_lang_attr = f"{rdf_attribute_prefix}lang" self.__rdf_datatype_attr = f"{rdf_attribute_prefix}datatype" + # Pre-compile regex patterns for container predicates + rdf_ns = str(RDF) + self.__container_pattern_n = re.compile(f"^{re.escape(rdf_ns)}_[0-9]+$") + self.__container_pattern_li = re.compile(f"^{re.escape(rdf_ns)}li$") + # An RDF to ArangoDB variable used as a buffer # to store the to-be-inserted ArangoDB documents (RDF-to-ArangoDB). self.__adb_docs: ADBDocs @@ -1035,6 +1040,12 @@ def rdf_to_arangodb_by_pgt( adb_import_kwargs, ) + ############################## + # PGT: Pre-compute RDF lists # + ############################## + + self.__precompute_rdf_list_info() + ########################### # PGT: Literal Statements # ########################### @@ -1073,8 +1084,8 @@ def rdf_to_arangodb_by_pgt( # Address the possibility of (s, p, o) being a part of the # structure of an RDF Collection or an RDF Container. - # TODO: Move out of loop, into a pre-processing step - rdf_list_col = self.__pgt_statement_is_part_of_rdf_list(s, p) + # Uses pre-computed RDF list information + rdf_list_col = self.__is_rdf_list_statement(s, p) if rdf_list_col: key = self.rdf_id_to_adb_label(str(p)) doc = self.__rdf_list_data[rdf_list_col][s] @@ -1122,6 +1133,47 @@ def rdf_to_arangodb_by_pgt( return self.__pgt_create_adb_graph(name) + def __precompute_rdf_list_info(self) -> None: + """Pre-compute RDF list information for optimization. + + Collects all RDF list subjects categorized by type once + to avoid repeated computation in processing loops. + """ + # Pre-compute collection subjects (RDF.first, RDF.rest) + self.__rdf_collection_subjects = set() + for s in self.__rdf_graph.subjects(RDF.first, None): + self.__rdf_collection_subjects.add(s) + for s in self.__rdf_graph.subjects(RDF.rest, None): + self.__rdf_collection_subjects.add(s) + + # Pre-compute container subjects (container predicates _1, li, etc.) + self.__rdf_container_subjects = set() + for s, p, o in self.__rdf_graph: + if isinstance(s, BNode): + p_str = str(p) + if self.__container_pattern_n.match(p_str) or self.__container_pattern_li.match(p_str): + self.__rdf_container_subjects.add(s) + + # Combined set for backward compatibility and general filtering + self.__rdf_list_subjects = self.__rdf_collection_subjects | self.__rdf_container_subjects + + def __is_rdf_list_statement(self, s: RDFTerm, p: URIRef) -> str: + """Returns the list type or empty string if not a list statement. + + :param s: The RDF Subject + :param p: The RDF Predicate + :return: The list type or empty string if not a list statement + """ + # O(1) lookups using pre-computed categorized sets + if s in self.__rdf_collection_subjects and p in {RDF.first, RDF.rest}: + return "_COLLECTION_BNODE" + + if s in self.__rdf_container_subjects: + # Already pre-computed as container subject, no need for regex + return "_CONTAINER_BNODE" + + return "" + def write_adb_col_statements( self, rdf_graph: RDFGraph, @@ -1716,7 +1768,7 @@ def __process_adb_vertex( term = self.__adb_doc_to_rdf_term(adb_v, v_col) self.__term_map[adb_v["_id"]] = term - if type(term) is Literal: + if isinstance(term, Literal): return term sg = URIRef(adb_v.get(self.__rdf_sub_graph_uri_attr, "")) or None @@ -1987,7 +2039,7 @@ def __adb_val_to_rdf_val( :type sg: URIRef | None """ - if type(val) is list: + if isinstance(val, list): if self.__list_conversion == "static": for v in val: self.__adb_val_to_rdf_val(col, s, p, v, sg) @@ -2016,7 +2068,7 @@ def __adb_val_to_rdf_val( val = json.dumps(val) self.__add_to_rdf_graph(s, p, Literal(val), sg) - elif type(val) is dict: + elif isinstance(val, dict): if self.__dict_conversion == "static": bnode = BNode() self.__add_to_rdf_graph(s, p, bnode, sg) @@ -2141,7 +2193,7 @@ def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: # TODO: Populate adb docs? Or uncessary? - elif type(t) is URIRef: + elif isinstance(t, URIRef): t_col = self.__URIREF_COL t_label = self.rdf_id_to_adb_label(t_str) @@ -2152,7 +2204,7 @@ def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: self.__rdf_type_attr: "URIRef", } - elif type(t) is BNode: + elif isinstance(t, BNode): t_col = self.__BNODE_COL self.__adb_docs[t_col][t_key] = { @@ -2161,7 +2213,7 @@ def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: self.__rdf_type_attr: "BNode", } - elif type(t) is Literal: + elif isinstance(t, Literal): t_col = self.__LITERAL_COL t_value = self.__get_literal_val(t, t_str) t_label = t_value @@ -2354,32 +2406,14 @@ def __pgt_parse_literal_statements( `ArangoRDF.__insert_adb_docs()`. :type adb_import_kwargs: Dict[str, Any] """ - # Pre-compute blacklist sets - rdf_list_subjects = set() - - # Collect RDF list subjects - for s in self.__rdf_graph.subjects(RDF.first, None): - rdf_list_subjects.add(s) - for s in self.__rdf_graph.subjects(RDF.rest, None): - rdf_list_subjects.add(s) - - # Use string patterns (same as your __pgt_statement_is_part_of_rdf_list method) - rdf_ns = str(RDF) - container_pattern_n = re.compile(f"^{re.escape(rdf_ns)}_[0-9]+$") - container_pattern_li = re.compile(f"^{re.escape(rdf_ns)}li$") - - # Direct iteration with efficient filtering literal_pairs = set() # Use set to avoid duplicates automatically for s, p, o in self.__rdf_graph: if not isinstance(o, Literal): continue - if s in rdf_list_subjects: - continue - - p_str = str(p) - if container_pattern_n.match(p_str) or container_pattern_li.match(p_str): + # Use pre-computed RDF list information (includes both collections and containers) + if s in self.__rdf_list_subjects: continue literal_pairs.add((s, p)) @@ -2402,22 +2436,15 @@ def __pgt_parse_literal_statements( else self.__rdf_graph.triples ) - # Optimization: track processed terms to avoid duplicate work - processed_terms = set() - with Live(Group(bar_progress, spinner_progress)): for i, (s, p) in enumerate(data, 1): bar_progress.update(bar_progress_task, advance=1) s_meta = self.__pgt_get_term_metadata(s) - # if s not in processed_terms: self.__pgt_process_rdf_term(s_meta) - # processed_terms.add(s) p_meta = self.__pgt_get_term_metadata(p) - # if p not in processed_terms: self.__pgt_process_rdf_term(p_meta) - # processed_terms.add(p) _, s_col, s_key, _ = s_meta _, _, _, p_label = p_meta @@ -2496,7 +2523,7 @@ def __pgt_get_term_metadata(self, t: Union[URIRef, BNode, Literal]) -> RDFTermMe :rtype: Tuple[URIRef | BNode | Literal, str, str, str] """ # Quick return for Literals (no caching needed) - if type(t) is Literal: + if isinstance(t, Literal): return t, "", "", "" # No other metadata needed # Check cache first @@ -2581,14 +2608,14 @@ def __pgt_rdf_val_to_adb_val( # This flag is only active in ArangoRDF.__pgt_process_rdf_lists() if process_val_as_serialized_list: - doc[key] += f"'{val}'," if type(val) is str else f"{val}," + doc[key] += f"'{val}'," if isinstance(val, str) else f"{val}," return prev_val = doc.get(key) if prev_val is None: doc[key] = val - elif type(prev_val) is list: + elif isinstance(prev_val, list): prev_val.append(val) else: doc[key] = [prev_val, val] @@ -2641,7 +2668,7 @@ def __pgt_process_rdf_term( "_to": _to, } - elif type(t) is URIRef: + elif isinstance(t, URIRef): self.__adb_docs[t_col][t_key] = { "_key": t_key, self.__rdf_uri_attr: str(t), @@ -2660,14 +2687,14 @@ def __pgt_process_rdf_term( self.__rdf_uri_attr: str(t), } - elif type(t) is BNode: + elif isinstance(t, BNode): self.__adb_docs[t_col][t_key] = { "_key": t_key, self.__rdf_label_attr: "", self.__rdf_type_attr: "BNode", } - elif type(t) is Literal and all([s_col, s_key, p_label]): + elif isinstance(t, Literal) and all([s_col, s_key, p_label]): self.__pgt_process_rdf_literal( t, s_col, s_key, p_label, sg_str, process_val_as_serialized_list ) @@ -2779,7 +2806,7 @@ def __pgt_process_statement( """ o, o_col, o_key, _ = o_meta - if type(o) is Literal or self.__pgt_object_is_head_of_rdf_list(o): + if isinstance(o, Literal) or self.__pgt_object_is_head_of_rdf_list(o): return _, s_col, s_key, _ = s_meta @@ -2827,54 +2854,12 @@ def __pgt_object_is_head_of_rdf_list(self, o: RDFTerm) -> bool: :return: Whether the object points to an RDF List or not. :rtype: bool """ - # TODO: Discuss repurcussions of this assumption - if type(o) is not BNode: + # Quick check: if not a BNode, it can't be an RDF list head + if not isinstance(o, BNode): return False - first = (o, RDF.first, None) - rest = (o, RDF.rest, None) - - if first in self.__rdf_graph or rest in self.__rdf_graph: - return True - - _n = (o, URIRef(f"{RDF}_1"), None) - li = (o, URIRef(f"{RDF}li"), None) - - if _n in self.__rdf_graph or li in self.__rdf_graph: - return True - - return False - - def __pgt_statement_is_part_of_rdf_list(self, s: RDFTerm, p: URIRef) -> str: - """RDF -> ArangoDB (PGT): Return the associated "Document Buffer" key - if the RDF Statement (s, p, _) is part of an RDF Collection or RDF Container - within the RDF Graph. Essential for unpacking the complicated data structure of - RDF Lists and re-building them as an ArangoDB Document Property. - - :param s: The RDF Subject. - :type s: URIRef | BNode - :param p: The RDF Predicate. - :type p: URIRef - :return: The **self.adb_docs** "Document Buffer" key associated - to the RDF Statement. If the statement is not part of an RDF - List, return an empty string. - :rtype: str - """ - # TODO: Discuss repurcussions of this assumption - if type(s) is not BNode: - return "" - - if p in {RDF.first, RDF.rest}: - return "_COLLECTION_BNODE" - - p_str = str(p) - _n = r"^http://www.w3.org/1999/02/22-rdf-syntax-ns#_[0-9]{1,}$" - li = r"^http://www.w3.org/1999/02/22-rdf-syntax-ns#li$" - - if re.match(_n, p_str) or re.match(li, p_str): - return "_CONTAINER_BNODE" - - return "" + # Use pre-computed RDF list subjects for O(1) lookup + return o in self.__rdf_list_subjects def __pgt_process_rdf_lists(self, bar_progress: Progress) -> None: """RDF -> ArangoDB (PGT): Process all RDF Collections & Containers @@ -3056,7 +3041,7 @@ def __pgt_unpack_rdf_container( # It is possible for the Container Membership Property # to be re-used in multiple statements (e.g rdf:li), # hence the reason why `value` can be a list or a single element. - value_as_list = value if type(value) is list else [value] + value_as_list = value if isinstance(value, list) else [value] for o in value_as_list: self.__pgt_process_rdf_list_object(doc, s_meta, p_meta, o, sg) @@ -3304,10 +3289,10 @@ def __get_subgraph_str(self, possible_sg: Optional[List[Any]]) -> str: sg = possible_sg[0] sg_identifier = sg.identifier if isinstance(sg, RDFGraph) else sg - if type(sg_identifier) is URIRef: + if isinstance(sg_identifier, URIRef): return str(sg_identifier) - if type(sg_identifier) is BNode: + if isinstance(sg_identifier, BNode): return "" # TODO: Revisit raise ValueError(f"Sub Graph Identifier is not a URIRef or BNode: {sg}") @@ -3832,7 +3817,7 @@ def __infer_and_introspect_dr( self.__e_col_map[e_col_type]["to"].add("Class") for t, t_col, t_key, t_label, dr_label in dr_meta: - if type(t) is Literal: + if isinstance(t, Literal): continue DR_COL = dr_label if is_pgt else self.__STATEMENT_COL From 5959c933a4e8faf7614555309845168451828231 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 18:30:50 -0400 Subject: [PATCH 08/20] remove: __term_metadata_cache --- arango_rdf/main.py | 25 +------------------------ 1 file changed, 1 insertion(+), 24 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 01015814..863c8c39 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -151,9 +151,6 @@ def __init__( # A mapping of Reified Subjects to their corresponding ArangoDB Edge. self.__reified_subject_map: Dict[Union[URIRef, BNode], Tuple[str, str, str]] - # Metadata cache for performance optimization - self.__term_metadata_cache: Dict[str, RDFTermMeta] = {} - # Commonly used URIs self.__rdfs_resource_str = str(RDFS.Resource) self.__rdfs_class_str = str(RDFS.Class) @@ -921,8 +918,6 @@ def rdf_to_arangodb_by_pgt( """ raise TypeError(m) - self.clear_term_metadata_cache() - namespace_prefixes = [] if namespace_collection_name: namespace_prefixes = [ @@ -1129,8 +1124,6 @@ def rdf_to_arangodb_by_pgt( logger.debug(result) - self.clear_term_metadata_cache() - return self.__pgt_create_adb_graph(name) def __precompute_rdf_list_info(self) -> None: @@ -2526,12 +2519,7 @@ def __pgt_get_term_metadata(self, t: Union[URIRef, BNode, Literal]) -> RDFTermMe if isinstance(t, Literal): return t, "", "", "" # No other metadata needed - # Check cache first t_str = str(t) - if t_str in self.__term_metadata_cache: - return self.__term_metadata_cache[t_str] - - # Compute metadata if not cached t_col = "" t_key = self.rdf_id_to_adb_key(t_str, t) t_label = self.rdf_id_to_adb_label(t_str) @@ -2559,18 +2547,7 @@ def __pgt_get_term_metadata(self, t: Union[URIRef, BNode, Literal]) -> RDFTermMe logger.debug(f"Found unknown resource: {t} ({t_key})") t_col = self.__UNKNOWN_RESOURCE - # Cache the result and return - result = t, str(t_col), t_key, t_label - self.__term_metadata_cache[t_str] = result - return result - - def clear_term_metadata_cache(self) -> None: - """Clear the metadata cache to free memory. - - Useful for long-running operations or when processing multiple - graphs to prevent excessive memory usage. - """ - self.__term_metadata_cache.clear() + return t, str(t_col), t_key, t_label def __pgt_rdf_val_to_adb_val( self, From 4b951f86c591bfbd2c8c2aaf52c734831541af29 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 20:51:45 -0400 Subject: [PATCH 09/20] mypy --- tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 506821c3..f7d7b2c5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -55,7 +55,7 @@ class NoTimeoutHTTPClient(DefaultHTTPClient): @pytest.fixture(autouse=True) -def reset_arango_db(): +def reset_arango_db() -> None: global db for g in db.graphs(): db.delete_graph(g["name"], drop_collections=True) From f1c1af70fa0e033237c67b49a2a00d88ffb87d90 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 20:51:56 -0400 Subject: [PATCH 10/20] new: `__pgt_parse_non_literal_statements` --- arango_rdf/main.py | 187 +++++++++++++++++++++++++++------------------ 1 file changed, 111 insertions(+), 76 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 863c8c39..6b0d1ccc 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -94,8 +94,8 @@ def __init__( self.insert_async = insert_async self.controller: ArangoRDFController = controller - self.controller.db: StandardDatabase = db - self.controller.async_db: AsyncDatabase = self.async_db + self.controller.db = db + self.controller.async_db = self.async_db # Set the RDF attribute prefix self.__rdf_attribute_prefix = rdf_attribute_prefix @@ -1042,65 +1042,55 @@ def rdf_to_arangodb_by_pgt( self.__precompute_rdf_list_info() ########################### - # PGT: Literal Statements # + # PGT: Prepare Statements # ########################### - self.__pgt_parse_literal_statements( - contextualize_statement_func, - batch_size, - adb_import_kwargs, - ) - - ############# - # PGT: Main # - ############# - - s: RDFTerm # Subject - p: URIRef # Predicate - o: RDFTerm # Object - - rdf_graph_size = len(self.__rdf_graph) - batch_size = batch_size or rdf_graph_size - bar_progress = get_bar_progress("(RDF → ADB): PGT", "#08479E") - bar_progress_task = bar_progress.add_task("", total=rdf_graph_size) - spinner_progress = get_import_spinner_progress(" ") - statements = ( self.__rdf_graph.quads if isinstance(self.__rdf_graph, RDFConjunctiveGraph) else self.__rdf_graph.triples ) - with Live(Group(bar_progress, spinner_progress)): - for i, (s, p, o, *sg) in enumerate(statements((None, None, None)), 1): - bar_progress.advance(bar_progress_task) - - logger.debug(f"PGT: {s} {p} {o} {sg}") + literal_statements = defaultdict(list) + non_literal_statements = defaultdict(list) + for s, p, o, *sg in statements((None, None, None)): + if ( + isinstance(o, Literal) + and s not in self.__rdf_collection_subjects + and s not in self.__rdf_container_subjects + ): + literal_statements[(s, p)].append((o, sg)) + else: + non_literal_statements[(s, p)].append((o, sg)) - # Address the possibility of (s, p, o) being a part of the - # structure of an RDF Collection or an RDF Container. - # Uses pre-computed RDF list information - rdf_list_col = self.__is_rdf_list_statement(s, p) - if rdf_list_col: - key = self.rdf_id_to_adb_label(str(p)) - doc = self.__rdf_list_data[rdf_list_col][s] - self.__pgt_rdf_val_to_adb_val(doc, key, o) - continue + ########################### + # PGT: Literal Statements # + ########################### - self.__pgt_process_subject_predicate_object( - s, p, o, sg, None, contextualize_statement_func - ) + self.__pgt_parse_literal_statements( + literal_statements, + contextualize_statement_func, + batch_size, + adb_import_kwargs, + ) - if i % batch_size == 0: - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + ############################### + # PGT: Non-Literal Statements # + ############################### - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__pgt_parse_non_literal_statements( + non_literal_statements, + contextualize_statement_func, + batch_size, + adb_import_kwargs, + ) ################## # PGT: RDF Lists # ################## - bar_progress = get_bar_progress("(RDF → ADB): PGT [RDF Lists]", "#EF7D00") + bar_progress = get_bar_progress("(RDF → ADB): PGT [Lists]", "#EF7D00") + spinner_progress = get_import_spinner_progress(" ") with Live(Group(bar_progress, spinner_progress)): self.__pgt_process_rdf_lists(bar_progress) self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) @@ -1138,18 +1128,17 @@ def __precompute_rdf_list_info(self) -> None: self.__rdf_collection_subjects.add(s) for s in self.__rdf_graph.subjects(RDF.rest, None): self.__rdf_collection_subjects.add(s) - + # Pre-compute container subjects (container predicates _1, li, etc.) self.__rdf_container_subjects = set() - for s, p, o in self.__rdf_graph: + for s, p, _ in self.__rdf_graph: if isinstance(s, BNode): p_str = str(p) - if self.__container_pattern_n.match(p_str) or self.__container_pattern_li.match(p_str): + container_pattern_n = self.__container_pattern_n.match(p_str) + container_pattern_li = self.__container_pattern_li.match(p_str) + if container_pattern_n or container_pattern_li: self.__rdf_container_subjects.add(s) - # Combined set for backward compatibility and general filtering - self.__rdf_list_subjects = self.__rdf_collection_subjects | self.__rdf_container_subjects - def __is_rdf_list_statement(self, s: RDFTerm, p: URIRef) -> str: """Returns the list type or empty string if not a list statement. @@ -1160,7 +1149,7 @@ def __is_rdf_list_statement(self, s: RDFTerm, p: URIRef) -> str: # O(1) lookups using pre-computed categorized sets if s in self.__rdf_collection_subjects and p in {RDF.first, RDF.rest}: return "_COLLECTION_BNODE" - + if s in self.__rdf_container_subjects: # Already pre-computed as container subject, no need for regex return "_CONTAINER_BNODE" @@ -2378,6 +2367,9 @@ def __pgt_remove_blacklisted_statements(self) -> None: def __pgt_parse_literal_statements( self, + literal_statements: DefaultDict[ + Tuple[RDFTerm, URIRef], List[Tuple[Literal, List[Any]]] + ], pgt_contextualize_statement_func: Callable[..., None], batch_size: Optional[int], adb_import_kwargs: Dict[str, Any], @@ -2399,40 +2391,23 @@ def __pgt_parse_literal_statements( `ArangoRDF.__insert_adb_docs()`. :type adb_import_kwargs: Dict[str, Any] """ - literal_pairs = set() # Use set to avoid duplicates automatically - - for s, p, o in self.__rdf_graph: - if not isinstance(o, Literal): - continue - - # Use pre-computed RDF list information (includes both collections and containers) - if s in self.__rdf_list_subjects: - continue - - literal_pairs.add((s, p)) - - data = list(literal_pairs) s: RDFTerm p: URIRef o: Literal - total = len(data) + total = len(literal_statements) batch_size = batch_size or total - bar_progress = get_bar_progress("(RDF → ADB): PGT [RDF Literals]", "#EF7D00") + bar_progress = get_bar_progress("(RDF → ADB): PGT [Literals]", "#EF7D00") bar_progress_task = bar_progress.add_task("", total=total) spinner_progress = get_import_spinner_progress(" ") - statements = ( - self.__rdf_graph.quads - if isinstance(self.__rdf_graph, RDFConjunctiveGraph) - else self.__rdf_graph.triples - ) - with Live(Group(bar_progress, spinner_progress)): - for i, (s, p) in enumerate(data, 1): + for i, (k, v) in enumerate(literal_statements.items(), 1): bar_progress.update(bar_progress_task, advance=1) + s, p = k + s_meta = self.__pgt_get_term_metadata(s) self.__pgt_process_rdf_term(s_meta) @@ -2442,7 +2417,7 @@ def __pgt_parse_literal_statements( _, s_col, s_key, _ = s_meta _, _, _, p_label = p_meta - for _, _, o, *sg in statements((s, p, None)): + for o, sg in v: sg_str = self.__get_subgraph_str(sg) o_meta = self.__pgt_get_term_metadata(o) @@ -2450,7 +2425,67 @@ def __pgt_parse_literal_statements( pgt_contextualize_statement_func(s_meta, p_meta, o_meta, sg_str) - self.__rdf_graph.remove((s, p, o)) + if i % batch_size == 0: + self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + + self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + + def __pgt_parse_non_literal_statements( + self, + non_literal_statements: DefaultDict[ + Tuple[RDFTerm, URIRef], List[Tuple[RDFTerm, List[Any]]] + ], + contextualize_statement_func: Callable[..., None], + batch_size: Optional[int], + adb_import_kwargs: Dict[str, Any], + ) -> None: + """RDF -> ArangoDB (PGT): Processes all non-literal RDF statements. + + :param non_literal_statements: Dictionary mapping (s,p) pairs to lists + of (o, sg) tuples for non-literal objects. + :type non_literal_statements: DefaultDict[ + Tuple[RDFTerm, URIRef], List[Tuple[RDFTerm, List[Any]]]] + ] + :param contextualize_statement_func: A function that contextualizes + an RDF Statement. A no-op function is used if Graph Contextualization + is disabled. + :type contextualize_statement_func: Callable[..., None] + :param batch_size: The batch size to use when inserting ArangoDB Documents. + Defaults to None. + :type batch_size: int | None + :param adb_import_kwargs: The keyword arguments to pass to + `ArangoRDF.__insert_adb_docs()`. + :type adb_import_kwargs: Dict[str, Any] + """ + + s: RDFTerm # Subject + p: URIRef # Predicate + o: RDFTerm # Object + + total = len(non_literal_statements) + batch_size = batch_size or total + bar_progress = get_bar_progress("(RDF → ADB): PGT [Non-Literals]", "#08479E") + bar_progress_task = bar_progress.add_task("", total=total) + spinner_progress = get_import_spinner_progress(" ") + + with Live(Group(bar_progress, spinner_progress)): + for i, (k, v) in enumerate(non_literal_statements.items(), 1): + bar_progress.update(bar_progress_task, advance=1) + + s, p = k + + rdf_list_col = self.__is_rdf_list_statement(s, p) + + for o, sg in v: + if rdf_list_col: + predicate_label = self.rdf_id_to_adb_label(str(p)) + doc = self.__rdf_list_data[rdf_list_col][s] + self.__pgt_rdf_val_to_adb_val(doc, predicate_label, o) + continue + + self.__pgt_process_subject_predicate_object( + s, p, o, sg, None, contextualize_statement_func + ) if i % batch_size == 0: self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) @@ -2836,7 +2871,7 @@ def __pgt_object_is_head_of_rdf_list(self, o: RDFTerm) -> bool: return False # Use pre-computed RDF list subjects for O(1) lookup - return o in self.__rdf_list_subjects + return o in self.__rdf_collection_subjects or o in self.__rdf_container_subjects def __pgt_process_rdf_lists(self, bar_progress: Progress) -> None: """RDF -> ArangoDB (PGT): Process all RDF Collections & Containers From e3b27c6acabcb35ce539cbbcfaf0ffa1a25401ad Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 21:14:54 -0400 Subject: [PATCH 11/20] remove: `self.__adb_docs` --- arango_rdf/main.py | 190 ++++++++++++++++++++++++++++----------------- 1 file changed, 118 insertions(+), 72 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 6b0d1ccc..ea41ff0e 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -114,10 +114,6 @@ def __init__( self.__container_pattern_n = re.compile(f"^{re.escape(rdf_ns)}_[0-9]+$") self.__container_pattern_li = re.compile(f"^{re.escape(rdf_ns)}li$") - # An RDF to ArangoDB variable used as a buffer - # to store the to-be-inserted ArangoDB documents (RDF-to-ArangoDB). - self.__adb_docs: ADBDocs - # Work-in-progress feature to enhance the Terminology Box of an RDF Graph # when importing to ArangoDB. self.__contextualize_graph = False @@ -697,8 +693,10 @@ def rdf_to_arangodb_by_rpt( self.__rdf_graph = rdf_graph self.__adb_key_statements = self.extract_adb_key_statements(rdf_graph) + # Create the ArangoDB documents buffer for this transformation + adb_docs: ADBDocs = defaultdict(lambda: defaultdict(dict)) + # Reset the ArangoDB Config - self.__adb_docs = defaultdict(lambda: defaultdict(dict)) self.__contextualize_graph = contextualize_graph self.__use_hashed_literals_as_keys = use_hashed_literals_as_keys @@ -726,7 +724,7 @@ def rdf_to_arangodb_by_rpt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = self.__rpt_contextualize_statement + contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__rpt_contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str) self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) @@ -746,6 +744,7 @@ def rdf_to_arangodb_by_rpt( self.__reified_subject_map = {} if flatten_reified_triples: self.__flatten_reified_triples( + adb_docs, self.__rpt_process_subject_predicate_object, contextualize_statement_func, batch_size, @@ -779,13 +778,13 @@ def rdf_to_arangodb_by_rpt( logger.debug(f"RPT: {s} {p} {o} {sg}") self.__rpt_process_subject_predicate_object( - s, p, o, sg, None, contextualize_statement_func + adb_docs, s, p, o, sg, None, contextualize_statement_func ) if i % batch_size == 0: - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) return self.__rpt_create_adb_graph(name) @@ -951,8 +950,10 @@ def rdf_to_arangodb_by_pgt( self.__predicate_collection = self.db.collection(predicate_collection_name) + # Create the ArangoDB documents buffer for this transformation + adb_docs: ADBDocs = defaultdict(lambda: defaultdict(dict)) + # Reset the ArangoDB Config - self.__adb_docs = defaultdict(lambda: defaultdict(dict)) self.__contextualize_graph = contextualize_graph # A unique set of instance variables to @@ -975,7 +976,7 @@ def rdf_to_arangodb_by_pgt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = self.__pgt_contextualize_statement + contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__pgt_contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str) self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) @@ -1029,6 +1030,7 @@ def rdf_to_arangodb_by_pgt( self.__reified_subject_map = {} if flatten_reified_triples: self.__flatten_reified_triples( + adb_docs, self.__pgt_process_subject_predicate_object, contextualize_statement_func, batch_size, @@ -1068,6 +1070,7 @@ def rdf_to_arangodb_by_pgt( ########################### self.__pgt_parse_literal_statements( + adb_docs, literal_statements, contextualize_statement_func, batch_size, @@ -1079,6 +1082,7 @@ def rdf_to_arangodb_by_pgt( ############################### self.__pgt_parse_non_literal_statements( + adb_docs, non_literal_statements, contextualize_statement_func, batch_size, @@ -1092,8 +1096,8 @@ def rdf_to_arangodb_by_pgt( bar_progress = get_bar_progress("(RDF → ADB): PGT [Lists]", "#EF7D00") spinner_progress = get_import_spinner_progress(" ") with Live(Group(bar_progress, spinner_progress)): - self.__pgt_process_rdf_lists(bar_progress) - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__pgt_process_rdf_lists(adb_docs, bar_progress) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) ########################### # PGT: Namespace Prefixes # @@ -2116,6 +2120,7 @@ def extract_adb_key_statements( def __rpt_process_subject_predicate_object( self, + adb_docs: ADBDocs, s: RDFTerm, p: URIRef, o: RDFTerm, @@ -2126,6 +2131,8 @@ def __rpt_process_subject_predicate_object( """RDF -> ArangoDB (RPT): Processes the RDF Statement (s, p, o) as an ArangoDB document for RPT. + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param s: The RDF Subject of the RDF Statement. :type s: URIRef | BNode :param p: The RDF Predicate of the RDF Statement. @@ -2145,19 +2152,21 @@ def __rpt_process_subject_predicate_object( """ sg_str = self.__get_subgraph_str(sg) - s_meta = self.__rpt_process_term(s) + s_meta = self.__rpt_process_term(adb_docs, s) - o_meta = self.__rpt_process_term(o) + o_meta = self.__rpt_process_term(adb_docs, o) - self.__rpt_process_statement(s_meta, p, o_meta, sg_str, reified_subject) + self.__rpt_process_statement(adb_docs, s_meta, p, o_meta, sg_str, reified_subject) contextualize_statement_func(s_meta, p, o_meta, sg_str) - def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: + def __rpt_process_term(self, adb_docs: ADBDocs, t: RDFTerm) -> RDFTermMeta: """RDF -> ArangoDB (RPT): Process an RDF Term as an ArangoDB document via RPT Standards. Returns the ArangoDB Collection & Document Key associated to the RDF term, along with its string representation. + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param t: The RDF Term to process :type t: URIRef | BNode | Literal :return: The RDF Term object, along with its associated ArangoDB @@ -2179,7 +2188,7 @@ def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: t_col = self.__URIREF_COL t_label = self.rdf_id_to_adb_label(t_str) - self.__adb_docs[t_col][t_key] = { + adb_docs[t_col][t_key] = { "_key": t_key, self.__rdf_uri_attr: t_str, self.__rdf_label_attr: t_label, @@ -2189,7 +2198,7 @@ def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: elif isinstance(t, BNode): t_col = self.__BNODE_COL - self.__adb_docs[t_col][t_key] = { + adb_docs[t_col][t_key] = { "_key": t_key, self.__rdf_label_attr: "", self.__rdf_type_attr: "BNode", @@ -2200,19 +2209,19 @@ def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: t_value = self.__get_literal_val(t, t_str) t_label = t_value - self.__adb_docs[t_col][t_key] = { + adb_docs[t_col][t_key] = { self.__rdf_value_attr: t_value, self.__rdf_label_attr: t_label, # TODO: REVISIT self.__rdf_type_attr: "Literal", } if self.__use_hashed_literals_as_keys: - self.__adb_docs[t_col][t_key]["_key"] = t_key + adb_docs[t_col][t_key]["_key"] = t_key if t.language: - self.__adb_docs[t_col][t_key][self.__rdf_lang_attr] = t.language + adb_docs[t_col][t_key][self.__rdf_lang_attr] = t.language elif t.datatype: - self.__adb_docs[t_col][t_key][self.__rdf_datatype_attr] = str( + adb_docs[t_col][t_key][self.__rdf_datatype_attr] = str( t.datatype ) @@ -2223,6 +2232,7 @@ def __rpt_process_term(self, t: RDFTerm) -> RDFTermMeta: def __rpt_process_statement( self, + adb_docs: ADBDocs, s_meta: RDFTermMeta, p: URIRef, o_meta: RDFTermMeta, @@ -2232,6 +2242,8 @@ def __rpt_process_statement( """RDF -> ArangoDB (RPT): Processes the RDF Statement (s, p, o) as an ArangoDB edge for RPT. + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param s_meta: The RDF Term Metadata associated to **s**. :type s_meta: arango_rdf.typings.RDFTermMeta :param p: The RDF Predicate URIRef of the statement (s, p, o). @@ -2263,6 +2275,7 @@ def __rpt_process_statement( e_key = self.hash(f"{s_key}-{p_key}-{o_key}") self.__add_adb_edge( + adb_docs, self.__STATEMENT_COL, e_key, _from, @@ -2273,10 +2286,12 @@ def __rpt_process_statement( ) def __rpt_contextualize_statement( - self, s_meta: RDFTermMeta, p: URIRef, o_meta: RDFTermMeta, sg_str: str + self, adb_docs: ADBDocs, s_meta: RDFTermMeta, p: URIRef, o_meta: RDFTermMeta, sg_str: str ) -> None: """RDF -> ArangoDB (RPT): Contextualizes the RDF Statement (s, p, o). + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param s_meta: The RDF Term Metadata associated to **s**. :type s_meta: arango_rdf.typings.RDFTermMeta :param p: The RDF Predicate URIRef of the statement (s, p, o). @@ -2287,8 +2302,8 @@ def __rpt_contextualize_statement( to this statement (if any). :type sg_str: str """ - p_meta = self.__rpt_process_term(p) - self.__contextualize_statement(s_meta, p_meta, o_meta, sg_str, is_pgt=False) + p_meta = self.__rpt_process_term(adb_docs, p) + self.__contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str, is_pgt=False) def __rpt_create_adb_graph(self, name: str) -> ADBGraph: """RDF -> ArangoDB (RPT): Create an ArangoDB graph based on @@ -2367,6 +2382,7 @@ def __pgt_remove_blacklisted_statements(self) -> None: def __pgt_parse_literal_statements( self, + adb_docs: ADBDocs, literal_statements: DefaultDict[ Tuple[RDFTerm, URIRef], List[Tuple[Literal, List[Any]]] ], @@ -2409,10 +2425,10 @@ def __pgt_parse_literal_statements( s, p = k s_meta = self.__pgt_get_term_metadata(s) - self.__pgt_process_rdf_term(s_meta) + self.__pgt_process_rdf_term(adb_docs, s_meta) p_meta = self.__pgt_get_term_metadata(p) - self.__pgt_process_rdf_term(p_meta) + self.__pgt_process_rdf_term(adb_docs, p_meta) _, s_col, s_key, _ = s_meta _, _, _, p_label = p_meta @@ -2421,17 +2437,18 @@ def __pgt_parse_literal_statements( sg_str = self.__get_subgraph_str(sg) o_meta = self.__pgt_get_term_metadata(o) - self.__pgt_process_rdf_literal(o, s_col, s_key, p_label, sg_str) + self.__pgt_process_rdf_literal(adb_docs, o, s_col, s_key, p_label, sg_str) pgt_contextualize_statement_func(s_meta, p_meta, o_meta, sg_str) if i % batch_size == 0: - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __pgt_parse_non_literal_statements( self, + adb_docs: ADBDocs, non_literal_statements: DefaultDict[ Tuple[RDFTerm, URIRef], List[Tuple[RDFTerm, List[Any]]] ], @@ -2484,16 +2501,17 @@ def __pgt_parse_non_literal_statements( continue self.__pgt_process_subject_predicate_object( - s, p, o, sg, None, contextualize_statement_func + adb_docs, s, p, o, sg, None, contextualize_statement_func ) if i % batch_size == 0: - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __pgt_process_subject_predicate_object( self, + adb_docs: ADBDocs, s: RDFTerm, p: URIRef, o: RDFTerm, @@ -2504,6 +2522,8 @@ def __pgt_process_subject_predicate_object( """RDF -> ArangoDB (PGT): Processes the RDF Statement (s, p, o) as an ArangoDB document for PGT. + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param s: The RDF Subject of the RDF Statement. :type s: URIRef | BNode :param p: The RDF Predicate of the RDF Statement. @@ -2524,15 +2544,15 @@ def __pgt_process_subject_predicate_object( sg_str = self.__get_subgraph_str(sg) s_meta = self.__pgt_get_term_metadata(s) - self.__pgt_process_rdf_term(s_meta) + self.__pgt_process_rdf_term(adb_docs, s_meta) p_meta = self.__pgt_get_term_metadata(p) - self.__pgt_process_rdf_term(p_meta) + self.__pgt_process_rdf_term(adb_docs, p_meta) o_meta = self.__pgt_get_term_metadata(o) - self.__pgt_process_object(s_meta, p_meta, o_meta, sg_str) + self.__pgt_process_object(adb_docs, s_meta, p_meta, o_meta, sg_str) - self.__pgt_process_statement(s_meta, p_meta, o_meta, sg_str, reified_subject) + self.__pgt_process_statement(adb_docs, s_meta, p_meta, o_meta, sg_str, reified_subject) contextualize_statement_func(s_meta, p_meta, o_meta, sg_str) @@ -2634,6 +2654,7 @@ def __pgt_rdf_val_to_adb_val( def __pgt_process_rdf_term( self, + adb_docs: ADBDocs, t_meta: RDFTermMeta, s_col: str = "", s_key: str = "", @@ -2665,7 +2686,7 @@ def __pgt_process_rdf_term( t, t_col, t_key, t_label = t_meta - if t_key in self.__adb_docs.get(t_col, {}): + if t_key in adb_docs.get(t_col, {}): return if t in self.__reified_subject_map: @@ -2674,14 +2695,14 @@ def __pgt_process_rdf_term( if self.__predicate_collection is not None: t_col = self.__predicate_collection.name - self.__adb_docs[t_col][t_key] = { + adb_docs[t_col][t_key] = { "_key": t_key, "_from": _from, "_to": _to, } elif isinstance(t, URIRef): - self.__adb_docs[t_col][t_key] = { + adb_docs[t_col][t_key] = { "_key": t_key, self.__rdf_uri_attr: str(t), self.__rdf_label_attr: t_label, @@ -2693,14 +2714,14 @@ def __pgt_process_rdf_term( and t_col != self.__UNKNOWN_RESOURCE ): uri_col = self.__uri_map_collection.name - self.__adb_docs[uri_col][t_key] = { + adb_docs[uri_col][t_key] = { "_key": t_key, "collection": t_col, self.__rdf_uri_attr: str(t), } elif isinstance(t, BNode): - self.__adb_docs[t_col][t_key] = { + adb_docs[t_col][t_key] = { "_key": t_key, self.__rdf_label_attr: "", self.__rdf_type_attr: "BNode", @@ -2708,7 +2729,7 @@ def __pgt_process_rdf_term( elif isinstance(t, Literal) and all([s_col, s_key, p_label]): self.__pgt_process_rdf_literal( - t, s_col, s_key, p_label, sg_str, process_val_as_serialized_list + adb_docs, t, s_col, s_key, p_label, sg_str, process_val_as_serialized_list ) else: @@ -2716,6 +2737,7 @@ def __pgt_process_rdf_term( def __pgt_process_rdf_literal( self, + adb_docs: ADBDocs, literal: Literal, s_col: str, s_key: str, @@ -2726,6 +2748,8 @@ def __pgt_process_rdf_literal( """RDF -> ArangoDB (PGT): Process an RDF Literal as an ArangoDB document property. + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param literal: The RDF Literal to process :type literal: Literal :param s_col: The ArangoDB document collection of the Subject associated @@ -2745,7 +2769,7 @@ def __pgt_process_rdf_literal( property. Defaults to False. :type process_val_as_serialized_list: bool """ - doc = self.__adb_docs[s_col][s_key] + doc = adb_docs[s_col][s_key] val = self.__get_literal_val(literal, str(literal)) self.__pgt_rdf_val_to_adb_val(doc, p_label, val, process_val_as_serialized_list) @@ -2753,7 +2777,7 @@ def __pgt_process_rdf_literal( doc[self.__rdf_sub_graph_uri_attr] = sg_str def __pgt_process_object( - self, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, sg_str: str + self, adb_docs: ADBDocs, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, sg_str: str ) -> None: """RDF -> ArangoDB (PGT): Processes the RDF Object into ArangoDB. Given the possibily of the RDF Object being used as the "root" of @@ -2761,6 +2785,8 @@ def __pgt_process_object( function is used to prevent calling `__pgt_process_rdf_term` if it is not required. + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param s_meta: The RDF Term Metadata associated to the RDF Subject of the statement containing the RDF Object. :type s_meta: arango_rdf.typings.RDFTermMeta @@ -2783,10 +2809,11 @@ def __pgt_process_object( self.__rdf_list_heads[s][p] = head else: - self.__pgt_process_rdf_term(o_meta, s_col, s_key, p_label, sg_str=sg_str) + self.__pgt_process_rdf_term(adb_docs, o_meta, s_col, s_key, p_label, sg_str=sg_str) def __pgt_process_statement( self, + adb_docs: ADBDocs, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, @@ -2800,6 +2827,8 @@ def __pgt_process_statement( 1) The RDF Object within the RDF Statement is not a Literal 2) The RDF Object is not the "root" node of an RDF List structure + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param s_meta: The RDF Term Metadata associated to the RDF Subject of the statement containing the RDF Object. :type s_meta: arango_rdf.typings.RDFTermMeta @@ -2842,6 +2871,7 @@ def __pgt_process_statement( e_key = self.hash(f"{s_key}-{p_key}-{o_key}") self.__add_adb_edge( + adb_docs, e_col, e_key, _from, @@ -2873,7 +2903,7 @@ def __pgt_object_is_head_of_rdf_list(self, o: RDFTerm) -> bool: # Use pre-computed RDF list subjects for O(1) lookup return o in self.__rdf_collection_subjects or o in self.__rdf_container_subjects - def __pgt_process_rdf_lists(self, bar_progress: Progress) -> None: + def __pgt_process_rdf_lists(self, adb_docs: ADBDocs, bar_progress: Progress) -> None: """RDF -> ArangoDB (PGT): Process all RDF Collections & Containers within the RDF Graph prior to inserting the documents into ArangoDB. @@ -2903,7 +2933,7 @@ def __pgt_process_rdf_lists(self, bar_progress: Progress) -> None: s_meta = self.__pgt_get_term_metadata(s) _, s_col, s_key, _ = s_meta - doc = self.__adb_docs[s_col][s_key] + doc = adb_docs[s_col][s_key] doc["_key"] = s_key for p, p_dict in s_dict.items(): @@ -2914,7 +2944,7 @@ def __pgt_process_rdf_lists(self, bar_progress: Progress) -> None: sg: str = p_dict["sub_graph"] doc[p_label] = "" - self.__pgt_process_rdf_list_object(doc, s_meta, p_meta, root, sg) + self.__pgt_process_rdf_list_object(adb_docs, doc, s_meta, p_meta, root, sg) doc[p_label] = doc[p_label].rstrip(",") # Delete doc[p_key] if there are no Literals within the RDF List @@ -2926,6 +2956,7 @@ def __pgt_process_rdf_lists(self, bar_progress: Progress) -> None: def __pgt_process_rdf_list_object( self, + adb_docs: ADBDocs, doc: Json, s_meta: RDFTermMeta, p_meta: RDFTermMeta, @@ -2963,7 +2994,7 @@ def __pgt_process_rdf_list_object( doc[p_label] += "[" next_bnode_dict = self.__rdf_list_data["_COLLECTION_BNODE"][o] - self.__pgt_unpack_rdf_collection(doc, s_meta, p_meta, next_bnode_dict, sg) + self.__pgt_unpack_rdf_collection(adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg) doc[p_label] = doc[p_label].rstrip(",") + "]," @@ -2971,7 +3002,7 @@ def __pgt_process_rdf_list_object( doc[p_label] += "[" next_bnode_dict = self.__rdf_list_data["_CONTAINER_BNODE"][o] - self.__pgt_unpack_rdf_container(doc, s_meta, p_meta, next_bnode_dict, sg) + self.__pgt_unpack_rdf_container(adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg) doc[p_label] = doc[p_label].rstrip(",") + "]," @@ -2981,13 +3012,14 @@ def __pgt_process_rdf_list_object( # Process the RDF Object as an ArangoDB Document self.__pgt_process_rdf_term( - o_meta, s_col, s_key, p_label, process_val_as_serialized_list=True + adb_docs, o_meta, s_col, s_key, p_label, process_val_as_serialized_list=True ) # Process the RDF Statement as an ArangoDB Edge - self.__pgt_process_statement(s_meta, p_meta, o_meta, sg) + self.__pgt_process_statement(adb_docs, s_meta, p_meta, o_meta, sg) def __pgt_unpack_rdf_collection( self, + adb_docs: ADBDocs, doc: Json, s_meta: RDFTermMeta, p_meta: RDFTermMeta, @@ -3013,16 +3045,17 @@ def __pgt_unpack_rdf_collection( """ first: RDFTerm = bnode_dict["first"] - self.__pgt_process_rdf_list_object(doc, s_meta, p_meta, first, sg) + self.__pgt_process_rdf_list_object(adb_docs, doc, s_meta, p_meta, first, sg) if "rest" in bnode_dict and bnode_dict["rest"] != RDF.nil: rest = bnode_dict["rest"] next_bnode_dict = self.__rdf_list_data["_COLLECTION_BNODE"][rest] - self.__pgt_unpack_rdf_collection(doc, s_meta, p_meta, next_bnode_dict, sg) + self.__pgt_unpack_rdf_collection(adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg) def __pgt_unpack_rdf_container( self, + adb_docs: ADBDocs, doc: Json, s_meta: RDFTermMeta, p_meta: RDFTermMeta, @@ -3055,13 +3088,15 @@ def __pgt_unpack_rdf_container( # hence the reason why `value` can be a list or a single element. value_as_list = value if isinstance(value, list) else [value] for o in value_as_list: - self.__pgt_process_rdf_list_object(doc, s_meta, p_meta, o, sg) + self.__pgt_process_rdf_list_object(adb_docs, doc, s_meta, p_meta, o, sg) def __pgt_contextualize_statement( - self, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, sg_str: str + self, adb_docs: ADBDocs, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, sg_str: str ) -> None: """RDF -> ArangoDB (PGT): Contextualizes the RDF Statement (s, p, o). + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param s_meta: The RDF Term Metadata associated to **s**. :type s_meta: arango_rdf.typings.RDFTermMeta :param p_meta: The RDF Term Metadata associated to **p**. @@ -3072,7 +3107,7 @@ def __pgt_contextualize_statement( to this statement (if any). :type sg_str: str """ - self.__contextualize_statement(s_meta, p_meta, o_meta, sg_str, is_pgt=True) + self.__contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str, is_pgt=True) def __pgt_create_adb_graph(self, name: str) -> ADBGraph: """RDF -> ArangoDB (PGT): Create an ArangoDB graph based @@ -3191,6 +3226,7 @@ def __load_meta_ontology(self, rdf_graph: RDFGraph) -> RDFConjunctiveGraph: def __flatten_reified_triples( self, + adb_docs: ADBDocs, process_subject_predicate_object: Callable[..., None], contextualize_statement_func: Callable[..., None], batch_size: Optional[int], @@ -3232,7 +3268,7 @@ def process_reified_subject( process_reified_subject(new_reified_subject, sg) process_subject_predicate_object( - s, p, o, sg, reified_subject, contextualize_statement_func + adb_docs, s, p, o, sg, reified_subject, contextualize_statement_func ) # Remove the reified triple from the RDF Graph @@ -3283,9 +3319,9 @@ def process_reified_subject( process_reified_subject(reified_subject, sg) if i % batch_size == 0: - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) - self.__insert_adb_docs(spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __get_subgraph_str(self, possible_sg: Optional[List[Any]]) -> str: """RDF -> ArangoDB: Extract the sub-graph URIRef string of a quad (if any). @@ -3311,6 +3347,7 @@ def __get_subgraph_str(self, possible_sg: Optional[List[Any]]) -> str: def __add_adb_edge( self, + adb_docs: ADBDocs, col: str, key: str, _from: str, @@ -3320,9 +3357,11 @@ def __add_adb_edge( _sg: str, ) -> None: """RDF -> ArangoDB: Insert the JSON-equivalent of an ArangoDB Edge - into `self.adb_docs` for temporary storage, until it gets + into `adb_docs` for temporary storage, until it gets ingested into the **col** ArangoDB Collection. + :param adb_docs: The ArangoDB documents buffer to populate. + :type adb_docs: ADBDocs :param col: The name of the ArangoDB Edge Collection. :type col: str :param key: The ArangoDB Key of the Edge. @@ -3342,8 +3381,8 @@ def __add_adb_edge( if self.__predicate_collection is not None: col = self.__predicate_collection.name - self.__adb_docs[col][key] = { - **self.__adb_docs[col][key], + adb_docs[col][key] = { + **adb_docs[col][key], "_key": key, "_from": _from, "_to": _to, @@ -3353,7 +3392,7 @@ def __add_adb_edge( } if _sg: - self.__adb_docs[col][key][self.__rdf_sub_graph_uri_attr] = _sg + adb_docs[col][key][self.__rdf_sub_graph_uri_attr] = _sg def __build_explicit_type_map( self, adb_adb_col_statement: Callable[..., None] = empty_func @@ -3670,10 +3709,12 @@ def __get_literal_val(self, t: Literal, t_str: str) -> Any: return t.value if t.value is not None else t_str def __insert_adb_docs( - self, spinner_progress: Progress, **adb_import_kwargs: Any + self, adb_docs: ADBDocs, spinner_progress: Progress, **adb_import_kwargs: Any ) -> None: """RDF -> ArangoDB: Insert ArangoDB documents into their ArangoDB collection. + :param adb_docs: The ArangoDB documents buffer to insert. + :type adb_docs: ADBDocs :param spinner_progress: The spinner progress bar. :type spinner_progress: rich.progress.Progress :param adb_import_kwargs: Keyword arguments to specify additional @@ -3681,7 +3722,7 @@ def __insert_adb_docs( https://docs.python-arango.com/en/main/specs.html#arango.collection.Collection.insert_many :param adb_import_kwargs: Any """ - if len(self.__adb_docs) == 0: + if len(adb_docs) == 0: return db = self.async_db if self.insert_async else self.db @@ -3693,10 +3734,10 @@ def __insert_adb_docs( adb_import_kwargs["raise_on_document_error"] = True # Avoiding "RuntimeError: dictionary changed size during iteration" - adb_cols = list(self.__adb_docs.keys()) + adb_cols = list(adb_docs.keys()) for col in adb_cols: - doc_list = self.__adb_docs[col].values() + doc_list = adb_docs[col].values() action = f"(RDF → ADB): Import '{col}' ({len(doc_list)})" spinner_progress_task = spinner_progress.add_task("", action=action) @@ -3717,13 +3758,14 @@ def __insert_adb_docs( logger.debug(f"Insert Result: {result}") - del self.__adb_docs[col] + del adb_docs[col] spinner_progress.stop_task(spinner_progress_task) spinner_progress.update(spinner_progress_task, visible=False) def __contextualize_statement( self, + adb_docs: ADBDocs, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, @@ -3757,6 +3799,7 @@ def __contextualize_statement( _to_col = "Class" if is_pgt else self.__URIREF_COL self.__add_adb_edge( + adb_docs, col=edge_col, key=self.hash(edge_key), _from=f"{_from_col}/{p_key}", @@ -3768,10 +3811,11 @@ def __contextualize_statement( # Run RDFS Domain/Range Inference & Introspection dr_meta = [(*s_meta, "domain"), (*o_meta, "range")] - self.__infer_and_introspect_dr(p, p_key, dr_meta, sg_str, is_pgt) + self.__infer_and_introspect_dr(adb_docs, p, p_key, dr_meta, sg_str, is_pgt) def __infer_and_introspect_dr( self, + adb_docs: ADBDocs, p: URIRef, p_key: str, dr_meta: List[Tuple[RDFTerm, str, str, str, str]], @@ -3842,6 +3886,7 @@ def __infer_and_introspect_dr( for _, class_key in self.__predicate_scope[p][dr_label]: key = self.hash(f"{t_key}-{self.__rdf_type_key}-{class_key}") self.__add_adb_edge( + adb_docs, col=TYPE_COL, key=key, _from=f"{t_col}/{t_key}", @@ -3868,6 +3913,7 @@ def __infer_and_introspect_dr( class_key = self.rdf_id_to_adb_key(class_str) key = self.hash(f"{p_key}-{dr_key}-{class_key}") self.__add_adb_edge( + adb_docs, col=DR_COL, key=key, _from=f"{P_COL}/{p_key}", From 9500a7fa6488de96a43b0cf4725de95e031d591a Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 22:10:05 -0400 Subject: [PATCH 12/20] Update main.py --- arango_rdf/main.py | 38 +++++++++++++++++++------------------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index ea41ff0e..22cdb117 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -960,6 +960,9 @@ def rdf_to_arangodb_by_pgt( # convert RDF Lists into JSON Lists during the PGT Process self.__rdf_list_heads: RDFListHeads = defaultdict(lambda: defaultdict(dict)) self.__rdf_list_data: RDFListData = defaultdict(lambda: defaultdict(dict)) + self.__rdf_list_subjects: Set[RDFTerm] = set() + self.__rdf_collection_subjects: Set[RDFTerm] = set() + self.__rdf_container_subjects: Set[RDFTerm] = set() # The ArangoDB Collection name of all unidentified RDF Resources self.__UNKNOWN_RESOURCE = f"{name}_UnknownResource" @@ -1056,11 +1059,7 @@ def rdf_to_arangodb_by_pgt( literal_statements = defaultdict(list) non_literal_statements = defaultdict(list) for s, p, o, *sg in statements((None, None, None)): - if ( - isinstance(o, Literal) - and s not in self.__rdf_collection_subjects - and s not in self.__rdf_container_subjects - ): + if isinstance(o, Literal) and s not in self.__rdf_list_subjects: literal_statements[(s, p)].append((o, sg)) else: non_literal_statements[(s, p)].append((o, sg)) @@ -1127,14 +1126,12 @@ def __precompute_rdf_list_info(self) -> None: to avoid repeated computation in processing loops. """ # Pre-compute collection subjects (RDF.first, RDF.rest) - self.__rdf_collection_subjects = set() for s in self.__rdf_graph.subjects(RDF.first, None): self.__rdf_collection_subjects.add(s) for s in self.__rdf_graph.subjects(RDF.rest, None): self.__rdf_collection_subjects.add(s) # Pre-compute container subjects (container predicates _1, li, etc.) - self.__rdf_container_subjects = set() for s, p, _ in self.__rdf_graph: if isinstance(s, BNode): p_str = str(p) @@ -1143,6 +1140,8 @@ def __precompute_rdf_list_info(self) -> None: if container_pattern_n or container_pattern_li: self.__rdf_container_subjects.add(s) + self.__rdf_list_subjects = self.__rdf_collection_subjects | self.__rdf_container_subjects + def __is_rdf_list_statement(self, s: RDFTerm, p: URIRef) -> str: """Returns the list type or empty string if not a list statement. @@ -1150,12 +1149,10 @@ def __is_rdf_list_statement(self, s: RDFTerm, p: URIRef) -> str: :param p: The RDF Predicate :return: The list type or empty string if not a list statement """ - # O(1) lookups using pre-computed categorized sets if s in self.__rdf_collection_subjects and p in {RDF.first, RDF.rest}: return "_COLLECTION_BNODE" if s in self.__rdf_container_subjects: - # Already pre-computed as container subject, no need for regex return "_CONTAINER_BNODE" return "" @@ -2493,17 +2490,20 @@ def __pgt_parse_non_literal_statements( rdf_list_col = self.__is_rdf_list_statement(s, p) - for o, sg in v: - if rdf_list_col: - predicate_label = self.rdf_id_to_adb_label(str(p)) - doc = self.__rdf_list_data[rdf_list_col][s] - self.__pgt_rdf_val_to_adb_val(doc, predicate_label, o) - continue + if rdf_list_col: + doc = self.__rdf_list_data[rdf_list_col][s] + predicate_label = self.rdf_id_to_adb_label(str(p)) - self.__pgt_process_subject_predicate_object( - adb_docs, s, p, o, sg, None, contextualize_statement_func - ) + for o, sg in v: + self.__pgt_rdf_val_to_adb_val(doc, predicate_label, o) + + else: + for o, sg in v: + self.__pgt_process_subject_predicate_object( + adb_docs, s, p, o, sg, None, contextualize_statement_func + ) + if i % batch_size == 0: self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) @@ -2901,7 +2901,7 @@ def __pgt_object_is_head_of_rdf_list(self, o: RDFTerm) -> bool: return False # Use pre-computed RDF list subjects for O(1) lookup - return o in self.__rdf_collection_subjects or o in self.__rdf_container_subjects + return o in self.__rdf_list_subjects def __pgt_process_rdf_lists(self, adb_docs: ADBDocs, bar_progress: Progress) -> None: """RDF -> ArangoDB (PGT): Process all RDF Collections & Containers From d7424b59ea5afc7cece4b8330f90c53ac96fe5f3 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Fri, 17 Oct 2025 22:10:51 -0400 Subject: [PATCH 13/20] fix: lint --- arango_rdf/main.py | 121 +++++++++++++++++++++++++++++++++------------ 1 file changed, 90 insertions(+), 31 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 22cdb117..e7774fca 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -695,7 +695,7 @@ def rdf_to_arangodb_by_rpt( # Create the ArangoDB documents buffer for this transformation adb_docs: ADBDocs = defaultdict(lambda: defaultdict(dict)) - + # Reset the ArangoDB Config self.__contextualize_graph = contextualize_graph self.__use_hashed_literals_as_keys = use_hashed_literals_as_keys @@ -724,7 +724,9 @@ def rdf_to_arangodb_by_rpt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__rpt_contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str) + contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__rpt_contextualize_statement( + adb_docs, s_meta, p_meta, o_meta, sg_str + ) self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) @@ -782,7 +784,9 @@ def rdf_to_arangodb_by_rpt( ) if i % batch_size == 0: - self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs( + adb_docs, spinner_progress, **adb_import_kwargs + ) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) @@ -952,7 +956,7 @@ def rdf_to_arangodb_by_pgt( # Create the ArangoDB documents buffer for this transformation adb_docs: ADBDocs = defaultdict(lambda: defaultdict(dict)) - + # Reset the ArangoDB Config self.__contextualize_graph = contextualize_graph @@ -979,7 +983,9 @@ def rdf_to_arangodb_by_pgt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__pgt_contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str) + contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__pgt_contextualize_statement( + adb_docs, s_meta, p_meta, o_meta, sg_str + ) self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) @@ -1140,7 +1146,9 @@ def __precompute_rdf_list_info(self) -> None: if container_pattern_n or container_pattern_li: self.__rdf_container_subjects.add(s) - self.__rdf_list_subjects = self.__rdf_collection_subjects | self.__rdf_container_subjects + self.__rdf_list_subjects = ( + self.__rdf_collection_subjects | self.__rdf_container_subjects + ) def __is_rdf_list_statement(self, s: RDFTerm, p: URIRef) -> str: """Returns the list type or empty string if not a list statement. @@ -2153,7 +2161,9 @@ def __rpt_process_subject_predicate_object( o_meta = self.__rpt_process_term(adb_docs, o) - self.__rpt_process_statement(adb_docs, s_meta, p, o_meta, sg_str, reified_subject) + self.__rpt_process_statement( + adb_docs, s_meta, p, o_meta, sg_str, reified_subject + ) contextualize_statement_func(s_meta, p, o_meta, sg_str) @@ -2218,9 +2228,7 @@ def __rpt_process_term(self, adb_docs: ADBDocs, t: RDFTerm) -> RDFTermMeta: if t.language: adb_docs[t_col][t_key][self.__rdf_lang_attr] = t.language elif t.datatype: - adb_docs[t_col][t_key][self.__rdf_datatype_attr] = str( - t.datatype - ) + adb_docs[t_col][t_key][self.__rdf_datatype_attr] = str(t.datatype) else: raise ValueError(f"Unable to process {t}") # pragma: no cover @@ -2283,7 +2291,12 @@ def __rpt_process_statement( ) def __rpt_contextualize_statement( - self, adb_docs: ADBDocs, s_meta: RDFTermMeta, p: URIRef, o_meta: RDFTermMeta, sg_str: str + self, + adb_docs: ADBDocs, + s_meta: RDFTermMeta, + p: URIRef, + o_meta: RDFTermMeta, + sg_str: str, ) -> None: """RDF -> ArangoDB (RPT): Contextualizes the RDF Statement (s, p, o). @@ -2300,7 +2313,9 @@ def __rpt_contextualize_statement( :type sg_str: str """ p_meta = self.__rpt_process_term(adb_docs, p) - self.__contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str, is_pgt=False) + self.__contextualize_statement( + adb_docs, s_meta, p_meta, o_meta, sg_str, is_pgt=False + ) def __rpt_create_adb_graph(self, name: str) -> ADBGraph: """RDF -> ArangoDB (RPT): Create an ArangoDB graph based on @@ -2434,12 +2449,16 @@ def __pgt_parse_literal_statements( sg_str = self.__get_subgraph_str(sg) o_meta = self.__pgt_get_term_metadata(o) - self.__pgt_process_rdf_literal(adb_docs, o, s_col, s_key, p_label, sg_str) + self.__pgt_process_rdf_literal( + adb_docs, o, s_col, s_key, p_label, sg_str + ) pgt_contextualize_statement_func(s_meta, p_meta, o_meta, sg_str) if i % batch_size == 0: - self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs( + adb_docs, spinner_progress, **adb_import_kwargs + ) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) @@ -2496,16 +2515,17 @@ def __pgt_parse_non_literal_statements( for o, sg in v: self.__pgt_rdf_val_to_adb_val(doc, predicate_label, o) - + else: for o, sg in v: - self.__pgt_process_subject_predicate_object( adb_docs, s, p, o, sg, None, contextualize_statement_func ) - + if i % batch_size == 0: - self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs( + adb_docs, spinner_progress, **adb_import_kwargs + ) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) @@ -2552,7 +2572,9 @@ def __pgt_process_subject_predicate_object( o_meta = self.__pgt_get_term_metadata(o) self.__pgt_process_object(adb_docs, s_meta, p_meta, o_meta, sg_str) - self.__pgt_process_statement(adb_docs, s_meta, p_meta, o_meta, sg_str, reified_subject) + self.__pgt_process_statement( + adb_docs, s_meta, p_meta, o_meta, sg_str, reified_subject + ) contextualize_statement_func(s_meta, p_meta, o_meta, sg_str) @@ -2729,7 +2751,13 @@ def __pgt_process_rdf_term( elif isinstance(t, Literal) and all([s_col, s_key, p_label]): self.__pgt_process_rdf_literal( - adb_docs, t, s_col, s_key, p_label, sg_str, process_val_as_serialized_list + adb_docs, + t, + s_col, + s_key, + p_label, + sg_str, + process_val_as_serialized_list, ) else: @@ -2777,7 +2805,12 @@ def __pgt_process_rdf_literal( doc[self.__rdf_sub_graph_uri_attr] = sg_str def __pgt_process_object( - self, adb_docs: ADBDocs, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, sg_str: str + self, + adb_docs: ADBDocs, + s_meta: RDFTermMeta, + p_meta: RDFTermMeta, + o_meta: RDFTermMeta, + sg_str: str, ) -> None: """RDF -> ArangoDB (PGT): Processes the RDF Object into ArangoDB. Given the possibily of the RDF Object being used as the "root" of @@ -2809,7 +2842,9 @@ def __pgt_process_object( self.__rdf_list_heads[s][p] = head else: - self.__pgt_process_rdf_term(adb_docs, o_meta, s_col, s_key, p_label, sg_str=sg_str) + self.__pgt_process_rdf_term( + adb_docs, o_meta, s_col, s_key, p_label, sg_str=sg_str + ) def __pgt_process_statement( self, @@ -2903,7 +2938,9 @@ def __pgt_object_is_head_of_rdf_list(self, o: RDFTerm) -> bool: # Use pre-computed RDF list subjects for O(1) lookup return o in self.__rdf_list_subjects - def __pgt_process_rdf_lists(self, adb_docs: ADBDocs, bar_progress: Progress) -> None: + def __pgt_process_rdf_lists( + self, adb_docs: ADBDocs, bar_progress: Progress + ) -> None: """RDF -> ArangoDB (PGT): Process all RDF Collections & Containers within the RDF Graph prior to inserting the documents into ArangoDB. @@ -2944,7 +2981,9 @@ def __pgt_process_rdf_lists(self, adb_docs: ADBDocs, bar_progress: Progress) -> sg: str = p_dict["sub_graph"] doc[p_label] = "" - self.__pgt_process_rdf_list_object(adb_docs, doc, s_meta, p_meta, root, sg) + self.__pgt_process_rdf_list_object( + adb_docs, doc, s_meta, p_meta, root, sg + ) doc[p_label] = doc[p_label].rstrip(",") # Delete doc[p_key] if there are no Literals within the RDF List @@ -2994,7 +3033,9 @@ def __pgt_process_rdf_list_object( doc[p_label] += "[" next_bnode_dict = self.__rdf_list_data["_COLLECTION_BNODE"][o] - self.__pgt_unpack_rdf_collection(adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg) + self.__pgt_unpack_rdf_collection( + adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg + ) doc[p_label] = doc[p_label].rstrip(",") + "]," @@ -3002,7 +3043,9 @@ def __pgt_process_rdf_list_object( doc[p_label] += "[" next_bnode_dict = self.__rdf_list_data["_CONTAINER_BNODE"][o] - self.__pgt_unpack_rdf_container(adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg) + self.__pgt_unpack_rdf_container( + adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg + ) doc[p_label] = doc[p_label].rstrip(",") + "]," @@ -3012,7 +3055,12 @@ def __pgt_process_rdf_list_object( # Process the RDF Object as an ArangoDB Document self.__pgt_process_rdf_term( - adb_docs, o_meta, s_col, s_key, p_label, process_val_as_serialized_list=True + adb_docs, + o_meta, + s_col, + s_key, + p_label, + process_val_as_serialized_list=True, ) # Process the RDF Statement as an ArangoDB Edge self.__pgt_process_statement(adb_docs, s_meta, p_meta, o_meta, sg) @@ -3051,7 +3099,9 @@ def __pgt_unpack_rdf_collection( rest = bnode_dict["rest"] next_bnode_dict = self.__rdf_list_data["_COLLECTION_BNODE"][rest] - self.__pgt_unpack_rdf_collection(adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg) + self.__pgt_unpack_rdf_collection( + adb_docs, doc, s_meta, p_meta, next_bnode_dict, sg + ) def __pgt_unpack_rdf_container( self, @@ -3091,7 +3141,12 @@ def __pgt_unpack_rdf_container( self.__pgt_process_rdf_list_object(adb_docs, doc, s_meta, p_meta, o, sg) def __pgt_contextualize_statement( - self, adb_docs: ADBDocs, s_meta: RDFTermMeta, p_meta: RDFTermMeta, o_meta: RDFTermMeta, sg_str: str + self, + adb_docs: ADBDocs, + s_meta: RDFTermMeta, + p_meta: RDFTermMeta, + o_meta: RDFTermMeta, + sg_str: str, ) -> None: """RDF -> ArangoDB (PGT): Contextualizes the RDF Statement (s, p, o). @@ -3107,7 +3162,9 @@ def __pgt_contextualize_statement( to this statement (if any). :type sg_str: str """ - self.__contextualize_statement(adb_docs, s_meta, p_meta, o_meta, sg_str, is_pgt=True) + self.__contextualize_statement( + adb_docs, s_meta, p_meta, o_meta, sg_str, is_pgt=True + ) def __pgt_create_adb_graph(self, name: str) -> ADBGraph: """RDF -> ArangoDB (PGT): Create an ArangoDB graph based @@ -3319,7 +3376,9 @@ def process_reified_subject( process_reified_subject(reified_subject, sg) if i % batch_size == 0: - self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) + self.__insert_adb_docs( + adb_docs, spinner_progress, **adb_import_kwargs + ) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) From 25ce810bb29206aecf8b11b4ab79716b0e331633 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Sat, 18 Oct 2025 22:52:57 -0400 Subject: [PATCH 14/20] new: `enable_pgt_cache` --- arango_rdf/main.py | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index e7774fca..e64ed41b 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -68,6 +68,11 @@ class ArangoRDF(AbstractArangoRDF): that using an underscore "_", results in these attributes being treated as ArangoDB system attributes. Using "$" is an alternative non-system prefix. :type rdf_attribute_prefix: str + :param insert_async: If True, will insert documents asynchronously. Defaults to False. + :type insert_async: bool + :param enable_pgt_cache: If True, will enable the PGT term metadata cache to avoid repeated computations. + Defaults to False. Not always useful, especially when terms are not repeated alot in the RDF graph. + :type enable_pgt_cache: bool :raise TypeError: On invalid parameter types """ @@ -78,6 +83,7 @@ def __init__( logging_lvl: Union[str, int] = logging.INFO, rdf_attribute_prefix: str = "_", insert_async: bool = False, + enable_pgt_cache: bool = False, ): self.set_logging(logging_lvl) @@ -131,6 +137,10 @@ def __init__( # e.g ( "4502") self.adb_key_uri = URIRef("http://www.arangodb.com/key") + # Cache for PGT term metadata to avoid repeated computations + self.enable_pgt_cache = enable_pgt_cache + self.pgt_term_metadata_cache: Dict[str, RDFTermMeta] = {} + # RDF Graph for maintaining the ArangoDB Collections & Keys # of the RDF Resources self.__adb_col_statements = RDFGraph() @@ -1117,7 +1127,9 @@ def rdf_to_arangodb_by_pgt( for prefix, uri in namespace_prefixes ] - result = self.db.collection(namespace_collection_name).insert_many( + db = self.db if self.insert_async else self.async_db + + result = db.collection(namespace_collection_name).insert_many( docs, overwrite=True, raise_on_document_error=True ) @@ -2597,6 +2609,11 @@ def __pgt_get_term_metadata(self, t: Union[URIRef, BNode, Literal]) -> RDFTermMe return t, "", "", "" # No other metadata needed t_str = str(t) + + if self.enable_pgt_cache: + if t_str in self.pgt_term_metadata_cache: + return self.pgt_term_metadata_cache[t_str] + t_col = "" t_key = self.rdf_id_to_adb_key(t_str, t) t_label = self.rdf_id_to_adb_label(t_str) @@ -2624,7 +2641,11 @@ def __pgt_get_term_metadata(self, t: Union[URIRef, BNode, Literal]) -> RDFTermMe logger.debug(f"Found unknown resource: {t} ({t_key})") t_col = self.__UNKNOWN_RESOURCE - return t, str(t_col), t_key, t_label + result = t, str(t_col), t_key, t_label + if self.enable_pgt_cache: + self.pgt_term_metadata_cache[t_str] = result + + return result def __pgt_rdf_val_to_adb_val( self, From 1d7fa2a1a4dcb5f69e6accd6e6c67cd7f354198c Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Sat, 18 Oct 2025 23:40:43 -0400 Subject: [PATCH 15/20] cleanup --- arango_rdf/main.py | 16 +++++++--------- tests/conftest.py | 2 +- 2 files changed, 8 insertions(+), 10 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index e64ed41b..16611eae 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -68,10 +68,12 @@ class ArangoRDF(AbstractArangoRDF): that using an underscore "_", results in these attributes being treated as ArangoDB system attributes. Using "$" is an alternative non-system prefix. :type rdf_attribute_prefix: str - :param insert_async: If True, will insert documents asynchronously. Defaults to False. + :param insert_async: If True, will insert documents asynchronously. + Defaults to False. :type insert_async: bool - :param enable_pgt_cache: If True, will enable the PGT term metadata cache to avoid repeated computations. - Defaults to False. Not always useful, especially when terms are not repeated alot in the RDF graph. + :param enable_pgt_cache: If True, will enable the PGT term metadata cache to avoid + repeated computations. Defaults to False. Not always useful, especially when + terms are not repeated alot in the RDF graph. :type enable_pgt_cache: bool :raise TypeError: On invalid parameter types """ @@ -734,9 +736,7 @@ def rdf_to_arangodb_by_rpt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__rpt_contextualize_statement( - adb_docs, s_meta, p_meta, o_meta, sg_str - ) + contextualize_statement_func = self.__rpt_contextualize_statement self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) @@ -993,9 +993,7 @@ def rdf_to_arangodb_by_pgt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = lambda s_meta, p_meta, o_meta, sg_str: self.__pgt_contextualize_statement( - adb_docs, s_meta, p_meta, o_meta, sg_str - ) + contextualize_statement_func = self.__pgt_contextualize_statement self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) diff --git a/tests/conftest.py b/tests/conftest.py index f7d7b2c5..8b6bdd13 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,9 +1,9 @@ import os -import pytest import subprocess from pathlib import Path from typing import Any, Dict, Set, Tuple +import pytest from arango import ArangoClient, DefaultHTTPClient from arango.database import StandardDatabase from rdflib import BNode From dd34022f38ee7c93760e86c6bd518370947fc34c Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Sat, 18 Oct 2025 23:43:30 -0400 Subject: [PATCH 16/20] fix: lint --- tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 8b6bdd13..b880c568 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -61,7 +61,7 @@ def reset_arango_db() -> None: db.delete_graph(g["name"], drop_collections=True) for c in db.collections(): - if c["system"] == False: + if c["system"] is False: db.delete_collection(c["name"]) From 3c05c25f62d95cf51f058d3ff38673ce68a66e13 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Sat, 18 Oct 2025 23:50:58 -0400 Subject: [PATCH 17/20] fix: lint --- arango_rdf/main.py | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 16611eae..7d2fde55 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -736,7 +736,16 @@ def rdf_to_arangodb_by_rpt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = self.__rpt_contextualize_statement + + def contextualize_statement_func( + s_meta: RDFTermMeta, + p_meta: RDFTermMeta, + o_meta: RDFTermMeta, + sg_str: str, + ) -> None: + return self.__rpt_contextualize_statement( + adb_docs, s_meta, p_meta, o_meta, sg_str + ) self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) @@ -993,7 +1002,16 @@ def rdf_to_arangodb_by_pgt( # NOTE: Graph Contextualization is an experimental work-in-progress contextualize_statement_func = empty_func if contextualize_graph: - contextualize_statement_func = self.__pgt_contextualize_statement + + def contextualize_statement_func( + s_meta: RDFTermMeta, + p_meta: RDFTermMeta, + o_meta: RDFTermMeta, + sg_str: str, + ) -> None: + return self.__pgt_contextualize_statement( + adb_docs, s_meta, p_meta, o_meta, sg_str + ) self.__rdf_graph = self.__load_meta_ontology(self.__rdf_graph) From 4c822ac65c9660d1ecb753a119885a7c7eec0c91 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Sat, 18 Oct 2025 23:57:16 -0400 Subject: [PATCH 18/20] new: progress bar for statement prep --- arango_rdf/main.py | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 7d2fde55..60191c3e 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -1090,11 +1090,21 @@ def contextualize_statement_func( literal_statements = defaultdict(list) non_literal_statements = defaultdict(list) - for s, p, o, *sg in statements((None, None, None)): - if isinstance(o, Literal) and s not in self.__rdf_list_subjects: - literal_statements[(s, p)].append((o, sg)) - else: - non_literal_statements[(s, p)].append((o, sg)) + + total = len(self.__rdf_graph) + bar_progress = get_bar_progress( + "(RDF → ADB): PGT [Prepare Statements]", "#EAD40B" + ) + bar_progress_task = bar_progress.add_task("", total=total) + + with Live(Group(bar_progress)): + for s, p, o, *sg in statements((None, None, None)): + bar_progress.update(bar_progress_task, advance=1) + + if isinstance(o, Literal) and s not in self.__rdf_list_subjects: + literal_statements[(s, p)].append((o, sg)) + else: + non_literal_statements[(s, p)].append((o, sg)) ########################### # PGT: Literal Statements # From 8de2b681c16bc08115f46ecb9975397b900b4b13 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Sun, 19 Oct 2025 15:28:21 -0400 Subject: [PATCH 19/20] `rich` cleanup --- arango_rdf/main.py | 32 +++++++++++++------------------- 1 file changed, 13 insertions(+), 19 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 60191c3e..058d2eb1 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -780,10 +780,10 @@ def contextualize_statement_func( p: URIRef # Predicate o: RDFTerm # Object - rdf_graph_size = len(self.__rdf_graph) - batch_size = batch_size or rdf_graph_size + total = len(self.__rdf_graph) + batch_size = batch_size or total bar_progress = get_bar_progress("(RDF → ADB): RPT", "#BF23C4") - bar_progress_task = bar_progress.add_task("", total=rdf_graph_size) + bar_progress_task = bar_progress.add_task("", total=total) spinner_progress = get_import_spinner_progress(" ") statements = ( @@ -794,8 +794,6 @@ def contextualize_statement_func( with Live(Group(bar_progress, spinner_progress)): for i, (s, p, o, *sg) in enumerate(statements((None, None, None)), 1): - bar_progress.advance(bar_progress_task) - logger.debug(f"RPT: {s} {p} {o} {sg}") self.__rpt_process_subject_predicate_object( @@ -803,10 +801,12 @@ def contextualize_statement_func( ) if i % batch_size == 0: + bar_progress.update(bar_progress_task, advance=batch_size) self.__insert_adb_docs( adb_docs, spinner_progress, **adb_import_kwargs ) + bar_progress.update(bar_progress_task, advance=total % batch_size) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) return self.__rpt_create_adb_graph(name) @@ -1091,16 +1091,10 @@ def contextualize_statement_func( literal_statements = defaultdict(list) non_literal_statements = defaultdict(list) - total = len(self.__rdf_graph) - bar_progress = get_bar_progress( - "(RDF → ADB): PGT [Prepare Statements]", "#EAD40B" - ) - bar_progress_task = bar_progress.add_task("", total=total) + with get_spinner_progress("(RDF → ADB): PGT [Prepare Statements]") as rp: + rp.add_task("") - with Live(Group(bar_progress)): for s, p, o, *sg in statements((None, None, None)): - bar_progress.update(bar_progress_task, advance=1) - if isinstance(o, Literal) and s not in self.__rdf_list_subjects: literal_statements[(s, p)].append((o, sg)) else: @@ -2470,8 +2464,6 @@ def __pgt_parse_literal_statements( with Live(Group(bar_progress, spinner_progress)): for i, (k, v) in enumerate(literal_statements.items(), 1): - bar_progress.update(bar_progress_task, advance=1) - s, p = k s_meta = self.__pgt_get_term_metadata(s) @@ -2494,10 +2486,12 @@ def __pgt_parse_literal_statements( pgt_contextualize_statement_func(s_meta, p_meta, o_meta, sg_str) if i % batch_size == 0: + bar_progress.update(bar_progress_task, advance=batch_size) self.__insert_adb_docs( adb_docs, spinner_progress, **adb_import_kwargs ) + bar_progress.update(bar_progress_task, advance=total % batch_size) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __pgt_parse_non_literal_statements( @@ -2541,8 +2535,6 @@ def __pgt_parse_non_literal_statements( with Live(Group(bar_progress, spinner_progress)): for i, (k, v) in enumerate(non_literal_statements.items(), 1): - bar_progress.update(bar_progress_task, advance=1) - s, p = k rdf_list_col = self.__is_rdf_list_statement(s, p) @@ -2561,10 +2553,12 @@ def __pgt_parse_non_literal_statements( ) if i % batch_size == 0: + bar_progress.update(bar_progress_task, advance=batch_size) self.__insert_adb_docs( adb_docs, spinner_progress, **adb_import_kwargs ) + bar_progress.update(bar_progress_task, advance=total % batch_size) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __pgt_process_subject_predicate_object( @@ -3415,18 +3409,18 @@ def process_reified_subject( with Live(Group(bar_progress, spinner_progress)): for i, (reified_subject, *sg) in enumerate(data, 1): - bar_progress.advance(bar_progress_task) - # Only process the reified triple if it has not been processed yet # i.e recursion if reified_subject not in self.__reified_subject_map: process_reified_subject(reified_subject, sg) if i % batch_size == 0: + bar_progress.update(bar_progress_task, advance=batch_size) self.__insert_adb_docs( adb_docs, spinner_progress, **adb_import_kwargs ) + bar_progress.update(bar_progress_task, advance=total % batch_size) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __get_subgraph_str(self, possible_sg: Optional[List[Any]]) -> str: From 6e821909be3874d7a251ca86068f80d8cbd45646 Mon Sep 17 00:00:00 2001 From: Anthony Mahanna Date: Sun, 19 Oct 2025 15:45:20 -0400 Subject: [PATCH 20/20] fix: modulo --- arango_rdf/main.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/arango_rdf/main.py b/arango_rdf/main.py index 058d2eb1..c16bb876 100644 --- a/arango_rdf/main.py +++ b/arango_rdf/main.py @@ -806,7 +806,8 @@ def contextualize_statement_func( adb_docs, spinner_progress, **adb_import_kwargs ) - bar_progress.update(bar_progress_task, advance=total % batch_size) + last_advance = total % batch_size if batch_size > 0 else 0 + bar_progress.update(bar_progress_task, advance=last_advance) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) return self.__rpt_create_adb_graph(name) @@ -2491,7 +2492,8 @@ def __pgt_parse_literal_statements( adb_docs, spinner_progress, **adb_import_kwargs ) - bar_progress.update(bar_progress_task, advance=total % batch_size) + last_advance = total % batch_size if batch_size > 0 else 0 + bar_progress.update(bar_progress_task, advance=last_advance) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __pgt_parse_non_literal_statements( @@ -2558,7 +2560,8 @@ def __pgt_parse_non_literal_statements( adb_docs, spinner_progress, **adb_import_kwargs ) - bar_progress.update(bar_progress_task, advance=total % batch_size) + last_advance = total % batch_size if batch_size > 0 else 0 + bar_progress.update(bar_progress_task, advance=last_advance) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __pgt_process_subject_predicate_object( @@ -3420,7 +3423,8 @@ def process_reified_subject( adb_docs, spinner_progress, **adb_import_kwargs ) - bar_progress.update(bar_progress_task, advance=total % batch_size) + last_advance = total % batch_size if batch_size > 0 else 0 + bar_progress.update(bar_progress_task, advance=last_advance) self.__insert_adb_docs(adb_docs, spinner_progress, **adb_import_kwargs) def __get_subgraph_str(self, possible_sg: Optional[List[Any]]) -> str: