init

Author: cmungall

src/rdf_sql_bulkloader/__init__.py

@@ -1,4 +1,6 @@
 """rdf-sql-bulkloader package."""
 from importlib import metadata
 
-__version__ = metadata.version(__name__)
+__version__ = metadata.version(__name__)
+
+from rdf_sql_bulkloader.loaders.sqlite3_bulkloader import SqliteBulkloader

src/rdf_sql_bulkloader/loaders/bulkloader.py

@@ -1,2 +1,44 @@
-class BulkLoader:
-    pass
+from abc import ABC
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Union, Tuple, Iterator
+
+import lightrdf
+
+DDL = """
+CREATE TABLE statement (
+	id TEXT,
+	subject TEXT,
+	predicate TEXT,
+	object TEXT,
+	value TEXT,
+	datatype TEXT,
+	language TEXT,
+        graph TEXT
+);
+"""
+
+URI = str
+SUBJECT = URI
+PREDICATE = URI
+OBJECT_URI = URI
+OBJECT_VALUE = str
+OBJECT_DATATYPE = str
+OBJECT_LANG = str
+
+STATEMENT = Tuple[SUBJECT, PREDICATE, OBJECT_URI, OBJECT_VALUE, OBJECT_DATATYPE, OBJECT_LANG]
+
+@dataclass
+class BulkLoader(ABC):
+    path: str
+
+    def bulkload(self, path: str):
+        raise NotImplemented
+
+    def statements(self, path: Union[Path, str]) -> Iterator[STATEMENT]:
+        doc = lightrdf.RDFDocument(str(path))
+        for t in doc.search_triples(None, None, None):
+            yield t
+
+    def ddl(self) -> str:
+        return DDL

src/rdf_sql_bulkloader/loaders/sqlite3_bulkloader.py

@@ -1,6 +1,15 @@
 import sqlite3
 
-from rdf_sql_bulkloader.main import BulkLoader
+from rdf_sql_bulkloader.loaders.bulkloader import BulkLoader
 
 
-class SqliteBulkloader(BulkLoader):
+class SqliteBulkloader(BulkLoader):
+
+    def bulkload(self, path: str):
+        con = sqlite3.connect(self.path)
+        con.execute(self.ddl())
+        tuples = []
+        for t in self.statements(path):
+            tuples.append(t)
+        print(len(tuples))
+        con.executemany("insert into statement(subject, predicate, object) values (?,?,?)", tuples)

tests/test_sqlite3_bulkloader.py

@@ -0,0 +1,14 @@
+"""Demo version test."""
+
+import unittest
+
+from rdf_sql_bulkloader import SqliteBulkloader
+from tests import TEST_INPUT_OWL
+
+
+class TestSqlit3BulkLoader(unittest.TestCase):
+    """Test sqlite3."""
+
+    def test_bulkload(self):
+        loader = SqliteBulkloader(":memory:")
+        loader.bulkload(TEST_INPUT_OWL)