cli-tidying

Author: cmungall

src/semsql/builder/cli.py

@@ -3,6 +3,7 @@
 import click
 from linkml_runtime import SchemaView
 from linkml_runtime.utils.formatutils import underscore
+from semsql.linkml import path_to_schema
 from sqlalchemy import text
 
 import semsql.builder.builder as builder
@@ -13,7 +14,7 @@
 @click.option("-v", "--verbose", count=True)
 @click.option("-q", "--quiet")
 def main(verbose: int, quiet: bool):
-    """Run the SSSOM CLI."""
+    """Run the SemSQL CLI."""
     if verbose >= 2:
         logging.basicConfig(level=logging.DEBUG)
     elif verbose == 1:
@@ -82,41 +83,40 @@ def query(input, query):
 
 
 @main.command()
-@click.argument("inputs", nargs=-1)
+@click.argument("views", nargs=-1)
 @click.option("--index/--no-index", default=True, help="Create indexes on each column")
 @click.option(
-    "--name",
-    "-n",
-    help="Name of class/view to materialize. If blank, will perform for ALL",
+    "--schema",
+    "-s",
+    help="Path o schema (optional)",
 )
-def view2table(inputs, name: str, index: bool):
+def view2table(views, schema, index: bool):
     """
     Generates a command that turns a view into a table
 
-    See https://github.com/cmungall/semantic-sql/issues/9
-
     Example usage:
-    ```
-    semsql view2table src/linkml/rdf.yaml -n rdfs_label_statement | sqlite3 db/pato.db
-    ```
+
+        semsql view2table rdfs_label_statement | sqlite3 db/pato.db
+
     """
-    for input in inputs:
-        sv = SchemaView(input)
-        for cn, c in sv.all_classes().items():
-            tn = underscore(cn)
-            if name is None or str(cn) == name or tn == name:
-                view = get_viewdef(c)
-                if view is not None:
-                    print(f"DROP VIEW {tn};")
-                    print(f"CREATE TABLE {tn} AS {view};")
-                    if index:
-                        for sn in sv.class_slots(cn):
-                            colname = underscore(sn)
-                            print(
-                                f"CREATE INDEX {tn}_{colname} ON {tn}({colname});"
-                            )
-                else:
-                    logging.error(f"No view for {cn}")
+    if not schema:
+        schema = str(path_to_schema())
+    sv = SchemaView(schema)
+    for cn, c in sv.all_classes().items():
+        tn = underscore(cn)
+        if not views or str(cn) in views or tn in views:
+            view = get_viewdef(c)
+            if view is not None:
+                print(f"DROP VIEW {tn};")
+                print(f"CREATE TABLE {tn} AS {view};")
+                if index:
+                    for sn in sv.class_slots(cn):
+                        colname = underscore(sn)
+                        print(
+                            f"CREATE INDEX {tn}_{colname} ON {tn}({colname});"
+                        )
+            else:
+                logging.error(f"No view for {cn}")
 
 
 if __name__ == "__main__":

src/semsql/linkml/__init__.py

@@ -0,0 +1,12 @@
+from pathlib import Path
+
+this_path = Path(__file__).parent
+
+def path_to_schema(schema_name: str = "semsql") -> Path:
+    """
+    Returns path to the name of a schema
+
+    :param schema_name: defaults to global schema (semsql)
+    :return:
+    """
+    return this_path / f"{schema_name}.yaml"

tests/test_builder/test_cli.py

@@ -29,6 +29,16 @@ def test_main_help(self):
         self.assertIn("make", out)
         self.assertIn("download", out)
         self.assertEqual(0, result.exit_code)
+        self.assertEqual(0, self.runner.invoke(main, ["download", "--help"]).exit_code)
+        self.assertEqual(0, self.runner.invoke(main, ["make", "--help"]).exit_code)
+        self.assertEqual(0, self.runner.invoke(main, ["query", "--help"]).exit_code)
+        self.assertEqual(0, self.runner.invoke(main, ["view2table", "--help"]).exit_code)
+
+    def test_view2table(self):
+        result = self.runner.invoke(main, ["view2table", TEST_DB])
+        out = result.stdout
+        err = result.stderr
+        self.assertEqual(0, result.exit_code)
 
     @unittest.skip("Requires Docker or installing dependencies")
     def test_make_db(self):