dependencies

Author: cmungall

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "rdf-sql-bulkloader"
-version = "0.1.0rc2"
+version = "0.0.0"
 description = "rdf-sql-bulkloader"
 authors = ["Chris Mungall <cjmungall@lbl.gov>"]
 license = "BSD 3"
@@ -13,8 +13,8 @@ tox = "^3.25.1"
 click = "^8.1.3"
 importlib = "^1.0.4"
 lightrdf = "^0.2.1"
-curies = "^0.2.0"
-prefixmaps = "^0.1.2"
+curies = "*"
+prefixmaps = "*"
 pyoxigraph = "^0.3.6"
 
 [tool.poetry.dev-dependencies]

pyproject.toml

@@ -31,9 +31,17 @@ def main(verbose: int, quiet: bool):
 
 @main.command()
 @click.option("--output", "-o", required=True)
+@click.option("--format",
+              "-f")
 @click.option(
     "--force/--no-force", default=False, show_default=True, help="Recreates db if already present"
 )
+@click.option(
+    "--append/--no-append", default=False, show_default=True, help="Append to existing db"
+)
+@click.option(
+    "--create-tables/--no-create-tables", default=True, show_default=True, help="Adds statements table"
+)
 @click.option(
     "--rdftab-compatibility/--no-rdftab-compatibility",
     default=True,
@@ -42,21 +50,24 @@ def main(verbose: int, quiet: bool):
 )
 @click.option("--named-prefix-map", "-P", multiple=True, help="Names of prefixmaps, e.g. obo")
 @click.argument("files", nargs=-1)
-def load_sqlite(files, output, force: bool, rdftab_compatibility: bool, named_prefix_map: tuple):
+def load_sqlite(files, format, output, append: bool, force: bool, rdftab_compatibility: bool, named_prefix_map: tuple, **kwargs):
     """Run the rdf-sql-bulkloader's demo command."""
     output_path = Path(output)
-    if output_path.exists():
-        if force:
-            output_path.unlink()
-        else:
-            raise ValueError(f"Path exists {output_path}")
+    if append:
+        if not output_path.exists():
+            raise ValueError(f"Cannot append as {output_path} does not exist")
+    else:
+        if output_path.exists():
+            if force:
+                output_path.unlink()
+            else:
+                raise ValueError(f"Path exists {output_path}")
     loader = SqliteBulkloader(
         output, named_prefix_maps=list(named_prefix_map) if named_prefix_map else None
     )
     loader.rdftab_compatibility = rdftab_compatibility
-    for file in files:
-        logging.info(f"Loading {file}")
-        loader.bulkload(file)
+    logging.info(f"Loading {files}")
+    loader.bulkload(list(files), format, **kwargs)
 
 
 if __name__ == "__main__":

src/rdf_sql_bulkloader/cli.py

@@ -28,6 +28,18 @@
 );
 """
 
+RDFTAB_STATEMENT_DDL = """
+CREATE TABLE statements (
+    stanza TEXT,
+    subject TEXT,
+    predicate TEXT,
+    object TEXT,
+    value TEXT,
+    datatype TEXT,
+    language TEXT
+);
+"""
+
 PREFIX_DDL = """
 CREATE TABLE prefix (
     prefix TEXT,
@@ -64,6 +76,9 @@ class BulkLoader(ABC):
     converter: Converter = None
     index_statements: bool = False
     rdftab_compatibility: bool = True
+    include_graph_name: bool = False
+    graph_name_from_ontology: bool = False
+    include_statement_id: bool = False
     use_shacl_namespaces: bool = True
     batch_size: int = field(default_factory=lambda: DEFAULT_CHUNK)
     _contract_uri_cache: Dict[Union[NamedNode, BlankNode], str] = field(default_factory=lambda: {})
@@ -111,6 +126,9 @@ def contract_uri(self, uri: Optional[NamedNode]) -> Optional[str]:
         else:
             return uri.value
 
+    def load_prefixes(self):
+        raise NotImplementedError
+
     def statements(self, path: Union[Path, str], mime_type=None) -> Iterator[STATEMENT]:
         """Yields statement rows from an RDF file."""
         if mime_type is None:
@@ -148,4 +166,7 @@ def statements(self, path: Union[Path, str], mime_type=None) -> Iterator[STATEME
 
     def ddl_statements(self) -> List[str]:
         """Return CREATE TABLE statements."""
-        return [STATEMENT_DDL, PREFIX_DDL]
+        if self.rdftab_compatibility:
+            return [RDFTAB_STATEMENT_DDL, PREFIX_DDL]
+        else:
+            return [STATEMENT_DDL, PREFIX_DDL]

src/rdf_sql_bulkloader/loaders/bulkloader.py

@@ -3,10 +3,13 @@
 import logging
 import sqlite3
 from dataclasses import dataclass
-from typing import Any, Iterable, List
+from typing import Any, Iterable, List, Union
 
 from rdf_sql_bulkloader.loaders.bulkloader import DEFAULT_CHUNK, BulkLoader
 
+
+logger = logging.getLogger(__name__)
+
 COLS = ["subject", "predicate", "object", "value", "datatype", "language"]
 
 RDFTAB_INSERT = """
@@ -45,7 +48,22 @@ class SqliteBulkloader(BulkLoader):
 
     connection: Any = None
 
-    def bulkload(self, path: str, mime_type=None):
+    def create_ddl(self):
+        """
+        Create DDL for a given path.
+
+        :return:
+        """
+        con = self.connection
+        for ddl_stmt in self.ddl_statements():
+            con.execute(ddl_stmt)
+        con.executemany(f"insert into prefix (prefix,base) values (?,?)", self.prefix_map.items())
+
+    def load_prefixes(self):
+        con = self.connection
+        raise NotImplementedError
+
+    def bulkload(self, paths: Union[str, List[str]], mime_type=None, create_tables=True):
         """
         Bulkloads from a path.
 
@@ -55,17 +73,28 @@ def bulkload(self, path: str, mime_type=None):
         """
         con = sqlite3.connect(self.database_path)
         self.connection = con
-        for ddl_stmt in self.ddl_statements():
-            con.execute(ddl_stmt)
-        con.executemany(f"insert into prefix (prefix,base) values (?,?)", self.prefix_map.items())
-        colstr = ",".join(COLS)
-        qs = ",".join(["?" for _ in COLS])
-        for chunk_it in chunk(self.statements(path, mime_type), self.batch_size):
-            tuples = []
-            for t in chunk_it:
-                tuples.append(t)
-            logging.info(f"Loaded {len(tuples)} loaded; {tuples[0]}")
-            con.executemany(f"insert into statement({colstr}) values ({qs})", tuples)
-        if self.rdftab_compatibility:
-            con.execute(RDFTAB_INSERT)
-        con.commit()
+        if create_tables:
+            self.create_ddl()
+        if not isinstance(paths, list):
+            paths = [paths]
+        for path in paths:
+            print(path)
+            logger.info(f"Loading {path} into {self.database_path} as {mime_type}...")
+            colstr = ",".join(COLS)
+            qs = ",".join(["?" for _ in COLS])
+            for chunk_it in chunk(self.statements(path, mime_type), self.batch_size):
+                tuples = []
+                for t in chunk_it:
+                    if self.rdftab_compatibility:
+                        tuples.append(t+(t[0],))
+                    else:
+                        tuples.append(t)
+                logging.info(f"Loaded {len(tuples)} loaded; {tuples[0]}")
+                if self.rdftab_compatibility:
+                    con.executemany(f"insert into statements({colstr},stanza) values ({qs},?)", tuples)
+                else:
+                    con.executemany(f"insert into statement({colstr}) values ({qs})", tuples)
+            con.commit()
+        #if self.rdftab_compatibility:
+        #    con.execute(RDFTAB_INSERT)
+

src/rdf_sql_bulkloader/loaders/sqlite3_bulkloader.py

@@ -95,13 +95,13 @@
 ]
 
 
-class TestSqlit3BulkLoader(unittest.TestCase):
+class TestSqlite3BulkLoader(unittest.TestCase):
     """Test sqlite3."""
 
     def test_bulkload(self):
         """Tests bulkload into an in-memory database."""
         loader = SqliteBulkloader(database_path=":memory:")
-        loader.rdftab_compatibility = True
+        loader.rdftab_compatibility = False
         loader.bulkload(TEST_INPUT_OWL)
         con = loader.connection
         cur = con.cursor()
@@ -111,6 +111,13 @@ def test_bulkload(self):
         #    print(f"S={s}")
         for case in CASES:
             self.assertIn(case, stmts)
+
+    def test_bulkload_compat(self):
+        loader = SqliteBulkloader(database_path=":memory:")
+        loader.rdftab_compatibility = True
+        loader.bulkload(TEST_INPUT_OWL)
+        con = loader.connection
+        cur = con.cursor()
         cur.execute("select * from statements WHERE subject=:subject", {"subject": NUCLEUS})
         stmts = list(cur.fetchall())
         # for s in stmts: