Add exclude path regex

Author: danlooo

pathtraits/cli.py

@@ -26,16 +26,21 @@ def main():
     default=DB_PATH,
     type=click.Path(file_okay=True, dir_okay=False),
 )
+@click.option(
+    "--exclude-regex",
+    default=None,
+)
 @click.option("-v", "--verbose", flag_value=True, default=False)
-def batch(path, db_path, verbose):
+def batch(path, db_path, exclude_regex, verbose):
     """
     Update database once, searches for all directories recursively.
 
     :param path: path to scan in batch mode recursively
     :param db_path: path to the database
+    :param exclude_regex: exclue file paths matching this regex
     :param verbose: enable verbose logging
     """
-    scan.batch(path, db_path, verbose)
+    scan.batch(path, db_path, exclude_regex, verbose)
 
 
 @main.command(help="Update database continiously, watches for new or changed files.")

pathtraits/db.py

@@ -74,7 +74,7 @@ def flatten_dict(dictionary: dict, root_key: str = "", separator: str = "/"):
         """
         items = []
         for key, value in dictionary.items():
-            new_key = root_key + separator + key if root_key else key
+            new_key = str(root_key) + str(separator) + str(key) if root_key else key
             if isinstance(value, MutableMapping):
                 items.extend(
                     TraitsDB.flatten_dict(value, new_key, separator=separator).items()
@@ -132,7 +132,7 @@ def get(self, table, cols="*", condition=None, **kwargs):
                 for (k, v) in kwargs.items()
             }
             condition = " AND ".join([f"{k}={v}" for (k, v) in escaped_kwargs.items()])
-        get_row_query = f"SELECT {cols} FROM {table} WHERE {condition};"
+        get_row_query = f"SELECT {cols} FROM [{table}] WHERE {condition};"
         response = self.execute(get_row_query)
 
         if response is None:
@@ -158,7 +158,7 @@ def put_path_id(self, path):
         get_row_query = f"SELECT id FROM path WHERE path = '{path}' LIMIT 1;"
         res = self.execute(get_row_query).fetchone()
         if res:
-            return res[0]
+            return res["id"]
         # create
         self.put("path", path=path)
         path_id = self.get("path", path=path, cols="id")["id"]
@@ -334,10 +334,9 @@ def add_pathpair(self, pair: PathPair):
 
                 # get element type for list
                 # add: handle lists with mixed element type
-                t = type(v[0]) if isinstance(v, list) else type(v)
+                t = type(v[0]) if isinstance(v, list) and len(v) > 0 else type(v)
                 k = f"{k}/{TraitsDB.sql_type(t)}"
                 if k not in self.traits:
-                    t = type(v[0]) if isinstance(v, list) else type(v)
                     self.create_trait_table(k, t)
                 if k in self.traits:
                     # add to list

pathtraits/scan.py

@@ -16,7 +16,7 @@
 
 
 # pylint: disable=W0102
-def scan_meta_yml(path, pathpairs=[]):
+def scan_meta_yml(path, pathpairs=[], exclude_regex=None):
     """
     Scan a directory recursively for meta yml files and return PathPairs.
 
@@ -27,8 +27,12 @@ def scan_meta_yml(path, pathpairs=[]):
         # faster than os.walk
         with os.scandir(path) as ents:
             for e in ents:
+                if not exclude_regex is None:
+                    if exclude_regex.search(e.path):
+                        logger.debug("exclude subtree path: %s", e.path)
+                        continue
                 if e.is_dir():
-                    scan_meta_yml(e.path, pathpairs)
+                    scan_meta_yml(e.path, pathpairs, exclude_regex)
                 else:
                     if not yaml_re.search(e.path):
                         continue
@@ -43,7 +47,7 @@ def scan_meta_yml(path, pathpairs=[]):
         return pathpairs
 
 
-def batch(path, db_path, verbose):
+def batch(path, db_path, exclude_regex, verbose):
     """
     Update database once, searches for all directories recursively.
 
@@ -57,7 +61,9 @@ def batch(path, db_path, verbose):
     if db_path is None:
         db_path = path + "/.pathtraits.db"
     db = TraitsDB(db_path)
-    pathpairs = scan_meta_yml(path)
+    if exclude_regex is not None:
+        exclude_regex = re.compile(exclude_regex)
+    pathpairs = scan_meta_yml(path, exclude_regex=exclude_regex)
     for pathpair in pathpairs:
         db.add_pathpair(pathpair)
 

test/test.py

@@ -19,7 +19,7 @@ class TestMain(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.db_path = tempfile.mkstemp()[1]
-        pathtraits.scan.batch("test/example", cls.db_path, False)
+        pathtraits.scan.batch("test/example", cls.db_path, "North_America$", False)
         cls.db = pathtraits.db.TraitsDB(cls.db_path)
 
     @classmethod
@@ -62,9 +62,17 @@ def test_example(self):
         for k, v in target.items():
             self.assertEqual(source[k], v)
 
+    def test_missing_north_america(self):
+        source = pathtraits.access.get_dict(
+            self.db, "test/example/Americas/North_America"
+        )
+        target = {"description": "all data", "is_example": True}
+        for k, v in target.items():
+            self.assertEqual(source[k], v)
+
     def test_data_view(self):
         source = len(self.db.execute("SELECT * FROM data;").fetchall())
-        target = 10
+        target = 8
         self.assertEqual(source, target)