satisfy flake8 - check_schema.py

Author: AhmetNSimsek

config_schema/check_schema.py

@@ -14,23 +14,21 @@
 
 use_thread = True
 
-skip_path = (
-    "snapshots/ebrainsquery/v1",
-    "venv"
-)
+skip_path = ("snapshots/ebrainsquery/v1", "venv")
 
 skip_types = []
 
+
 class ValidationResult(Enum):
-    SKIPPED="SKIPPED"
-    PASSED="PASSED"
-    FAILED="FAILED"
+    SKIPPED = "SKIPPED"
+    PASSED = "PASSED"
+    FAILED = "FAILED"
+
 
 BASE_URI = "http://example.com"
 
-ROOT_DIR = os.path.abspath(
-    f"{os.path.dirname(os.path.realpath(__file__))}/.."
-)
+ROOT_DIR = os.path.abspath(f"{os.path.dirname(os.path.realpath(__file__))}/..")
+
 
 def get_ref(schema):
     resolver = RefResolver(base_uri=BASE_URI, referrer=schema)
@@ -40,7 +38,6 @@ def get_ref(schema):
             if not filename.endswith(".json"):
                 continue
             with open(f"{dirpath}/{filename}", "r") as fp:
-
                 relative_path = dirpath.replace(walk_path, "") + "/" + filename
 
                 key0 = urljoin(BASE_URI, f"config_schema/{relative_path}")
@@ -50,9 +47,10 @@ def get_ref(schema):
 
                 resolver.store[key0] = schema
                 resolver.store[key1] = schema
-            
+
     return resolver
 
+
 def validate_json(path_to_json, fail_fast=False):
     if any([path_fragment in path_to_json for path_fragment in skip_path]):
         return (
@@ -65,43 +63,22 @@ def validate_json(path_to_json, fail_fast=False):
 
     # skip list
     if isinstance(json_obj, list):
-        return (
-            path_to_json,
-            ValidationResult.SKIPPED,
-            None
-        )
+        return (path_to_json, ValidationResult.SKIPPED, None)
     _type = json_obj.get("@type", None)
     if not _type:
         # TODO consolidate how error are raied
         if fail_fast:
             raise ValidationError(f"type does not exist: {path_to_json}")
-        return (
-            path_to_json,
-            ValidationResult.FAILED,
-            None
-        )
-    
+        return (path_to_json, ValidationResult.FAILED, None)
+
     # assert _schema is None
     if not _type or not _type.startswith("siibra"):
-        return (
-            path_to_json,
-            ValidationResult.SKIPPED,
-            None
-        )
+        return (path_to_json, ValidationResult.SKIPPED, None)
     if _type in skip_types:
-        return (
-            path_to_json,
-            ValidationResult.SKIPPED,
-            None
-        )
-    abspath = os.path.join(
-                ROOT_DIR, "config_schema", (_type + ".json")
-            )
+        return (path_to_json, ValidationResult.SKIPPED, None)
+    abspath = os.path.join(ROOT_DIR, "config_schema", (_type + ".json"))
     path_to_schema = os.path.abspath(abspath)
-    with open(
-        path_to_schema,
-        "r"
-    ) as fp:
+    with open(path_to_schema, "r") as fp:
         schema = json.load(fp)
     try:
         resolver = get_ref(schema)
@@ -110,52 +87,48 @@ def validate_json(path_to_json, fail_fast=False):
         if fail_fast:
             # TODO consolidate how error are raied
             raise e
-        return (
-            path_to_json,
-            ValidationResult.FAILED,
-            e
-        )
-    return (
-        path_to_json,
-        ValidationResult.PASSED,
-        None
-    )
-    
+        return (path_to_json, ValidationResult.FAILED, e)
+    return (path_to_json, ValidationResult.PASSED, None)
+
 
 def main(path_to_configuration: str, *args):
-    json_files = [f"{dirpath}/{filename}"
-                    for dirpath, dirnames, filenames in os.walk(path_to_configuration)
-                    for filename in filenames
-                    if filename.endswith(".json")
-                ]
+    json_files = [
+        f"{dirpath}/{filename}"
+        for dirpath, dirnames, filenames in os.walk(path_to_configuration)
+        for filename in filenames
+        if filename.endswith(".json")
+    ]
     # TODO use argparse
     fail_fast = "--fail-fast" in args
     if use_thread:
-        
-        with ThreadPoolExecutor(max_workers=4) as executor:
-            result = [progress for progress in tqdm(
-                executor.map(
-                    validate_json,
-                    json_files,
-                    repeat(fail_fast)
-                ),
-                total=len(json_files)
-            )]
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            result = [
+                progress
+                for progress in tqdm(
+                    executor.map(validate_json, json_files, repeat(fail_fast)),
+                    total=len(json_files),
+                )
+            ]
     else:
         result = [validate_json(f, fail_fast) for f in json_files]
 
     passed = [r for r in result if r[1] == ValidationResult.PASSED]
     failed = [r for r in result if r[1] == ValidationResult.FAILED]
     skipped = [r for r in result if r[1] == ValidationResult.SKIPPED]
-    print(f"Validation results: PASSED: {len(passed)} SKIPPED: {len(skipped)} FAILED: {len(failed)}")
+    print(
+        f"Validation results: PASSED: {len(passed)} SKIPPED: {len(skipped)} FAILED: {len(failed)}"
+    )
 
     if len(failed) > 0:
         print(failed)
         # TODO consolidate how error are raied
-        raise ValidationError(message="\n-----\n".join([f"{f[0]}: {str(f[2])}" for f in failed]))
+        raise ValidationError(
+            message="\n-----\n".join([f"{f[0]}: {str(f[2])}" for f in failed])
+        )
+
 
 if __name__ == "__main__":
     args = sys.argv[1:]
     if len(args) == 0:
-        raise RuntimeError(f"Need path to configuration directory")
+        raise RuntimeError("Need path to configuration directory")
     main(*args)