add: skip flag on commands

Author: dmartmillan

openvariant/commands/openvar.py

@@ -24,9 +24,10 @@ def openvar():
               help="Annotation path. eg: /path/annotation_vcf.yaml")
 @click.option('--header', is_flag=True, help="Show the result header.")
 @click.option('--output', '-o', default=None, help="File to write the output.")
-def cat(input_path: str, where: str or None, annotations: str or None, header: bool, output: str or None):
+@click.option('--skip', is_flag=True, help="Skip files and directories that are unreadable.")
+def cat(input_path: str, where: str or None, annotations: str or None, header: bool, output: str or None, skip: bool):
     """Print the parsed files on the stdout/"output"."""
-    cat_task(input_path, annotations, where, header, output)
+    cat_task(input_path, annotations, where, header, output, skip)
 
 
 @openvar.command(name="count", short_help='Number of rows that matches a specified criterion.')
@@ -38,10 +39,11 @@ def cat(input_path: str, where: str or None, annotations: str or None, header: b
 @click.option('--cores', '-c', type=click.INT, default=cpu_count(), help='Maximum processes to run in parallel.')
 @click.option('--quite', '-q', is_flag=True, help="Don't show the progress.")
 @click.option('--output', '-o', default=None, help="File to write the output.")
+@click.option('--skip', is_flag=True, help="Skip files and directories that are unreadable.")
 def count(input_path: str, where: str, group_by: str, cores: int, quite: bool, annotations: str or None,
-          output: str or None) -> None:
+          output: str or None, skip: bool) -> None:
     """Print on the stdout/"output" the number of rows that meets the criteria."""
-    result = count_task(input_path, annotations, group_by=group_by, where=where, cores=cores, quite=quite)
+    result = count_task(input_path, annotations, group_by=group_by, where=where, cores=cores, quite=quite, skip_files=skip)
     out_file = None
     if output:
         out_file = open(output, "w")
@@ -74,14 +76,15 @@ def count(input_path: str, where: str, group_by: str, cores: int, quite: bool, a
 @click.option('--cores', '-c', type=click.INT, default=cpu_count(), help='Maximum processes to run in parallel.')
 @click.option('--quite', '-q', is_flag=True, help="Don't show the progress.")
 @click.option('--output', '-o', help="File to write the output.", default=None)
+@click.option('--skip', is_flag=True, help="Skip files and directories that are unreadable.")
 def groupby(input_path: str, script: str, where: str, group_by: str, cores: int, quite: bool, annotations: str or None,
-            header: bool, show: bool, output: str or None):
+            header: bool, show: bool, output: str or None, skip: bool):
     """Print on the stdout/"output" the parsed files group by a specified field."""
     out_file = None
     if output:
         out_file = open(output, 'w')
     for group_key, group_result, command in group_by_task(input_path, annotations, script, key_by=group_by, where=where,
-                                                          cores=cores, quite=quite, header=header):
+                                                          cores=cores, quite=quite, header=header, skip_files=skip):
         for r in group_result:
             if command:
                 if output:

openvariant/find_files/find_files.py

@@ -7,6 +7,7 @@
 from os import listdir
 from os.path import isfile, join, isdir, dirname
 from typing import Generator
+import warnings
 
 from openvariant.annotation.annotation import Annotation
 from openvariant.annotation.config_annotation import ANNOTATION_EXTENSION
@@ -23,7 +24,7 @@ def _get_annotation(file_path, annotation):
             raise AttributeError("Unable to parse annotation file, check its location.")
 
 
-def _scan_files(base_path: str, annotation: Annotation, fix: bool):
+def _scan_files(base_path: str, annotation: Annotation, fix: bool, skip_files: bool):
     """Recursive exploration from a base path"""
     if isdir(base_path):
         if not fix:
@@ -34,22 +35,32 @@ def _scan_files(base_path: str, annotation: Annotation, fix: bool):
         for file_name in list_files:
             file_path = join(base_path, file_name)
             try:
-                for f, a in _scan_files(file_path, annotation, fix):
+                for f, a in _scan_files(file_path, annotation, fix, skip_files):
                     yield f, a
-            except PermissionError as e:
-                raise PermissionError(f"Unable to open {file_name}: {e}")
+            except PermissionError:
+                if skip_files:
+                    warnings.warn(f"Permission denied on {file_path}", UserWarning)
+                else:
+                    raise PermissionError(f"Permission denied on {file_path}")
     elif isfile(base_path):
         file_path = base_path
         try:
             for f, a in _get_annotation(file_path, annotation):
                 yield f, a
-        except PermissionError as e:
-            raise PermissionError(f"Unable to open {base_path}: {e}")
+        except PermissionError:
+            if skip_files:
+                warnings.warn(f"Permission denied on {file_path}", UserWarning)
+            else:
+                raise PermissionError(f"Permission denied on {file_path}")
     else:
-        raise FileNotFoundError(f"Unable to open {base_path}, it's not a file nor a directory.")
+        if skip_files:
+            warnings.warn(f"Unable to open {base_path}, it's not a file nor a directory.", UserWarning)
+        else:
+            raise PermissionError(f"Unable to open {base_path}, it's not a file nor a directory.")
+
 
 
-def _find_files(base_path: str, annotation: Annotation or None, fix: bool) -> Generator[str, Annotation, None]:
+def _find_files(base_path: str, annotation: Annotation or None, fix: bool, skip_files: bool) -> Generator[str, Annotation, None]:
     """Recursive exploration from a base path distinct if there's a fix annotation or no"""
     if not fix:
         if isfile(base_path):
@@ -59,11 +70,11 @@ def _find_files(base_path: str, annotation: Annotation or None, fix: bool) -> Ge
         for annotation_file in glob.iglob(join(annotation_path, "*.{}".format(ANNOTATION_EXTENSION))):
             annotation = Annotation(annotation_file)
 
-    for f, a in _scan_files(base_path, annotation, fix):
+    for f, a in _scan_files(base_path, annotation, fix, skip_files):
         yield f, a
 
 
-def findfiles(base_path: str, annotation_path: str or None = None) -> Generator[str, Annotation, None]:
+def findfiles(base_path: str, annotation_path: str or None = None, skip_files: bool = False) -> Generator[str, Annotation, None]:
     """Get each file and its proper annotation object.
 
     Parameters
@@ -72,6 +83,8 @@ def findfiles(base_path: str, annotation_path: str or None = None) -> Generator[
         Base path of input folder/file.
     annotation_path : str or None
         Path of annotation file.
+    skip_files : bool
+        Skip unreadable files and directories.
 
     Yields
     -------
@@ -81,5 +94,5 @@ def findfiles(base_path: str, annotation_path: str or None = None) -> Generator[
         The proper schema of each input file.
     """
     annotation, fix = (Annotation(annotation_path), True) if annotation_path is not None else (None, False)
-    for f, a in _find_files(base_path, annotation, fix):
+    for f, a in _find_files(base_path, annotation, fix, skip_files):
         yield f, a

openvariant/tasks/cat.py

@@ -16,7 +16,7 @@ def _format_line(line: List[str], out_format: str) -> str:
 
 
 def cat(base_path: str, annotation_path: str or None = None, where: str = None, header_show: bool = True,
-        output: str or None = None) -> None:
+        output: str or None = None, skip_files: bool = False) -> None:
     """Print on the stdout/"output" the parsed files.
 
     It will parse the input files with its proper annotation schema, and it'll show the result on the stdout.
@@ -34,12 +34,14 @@ def cat(base_path: str, annotation_path: str or None = None, where: str = None,
         Shows header on the output.
     output : str or None
         Save output on a file.
+    skip_files : bool
+        Skip unreadable files and directories.
     """
     out_file = None
     if output:
         out_file = open(output, "w")
-    for file, annotation in findfiles(base_path, annotation_path):
-        result = Variant(file, annotation)
+    for file, annotation in findfiles(base_path, annotation_path, skip_files):
+        result = Variant(file, annotation, skip_files)
         header = result.header
         if header_show:
             if output:

openvariant/tasks/count.py

@@ -16,13 +16,13 @@
 from openvariant.variant.variant import Variant
 
 
-def _count_task(selection: [str, str], group_by: str, where: str) -> Tuple[int, Union[dict, None]]:
+def _count_task(selection: [str, str], group_by: str, where: str, skip_files: bool) -> Tuple[int, Union[dict, None]]:
     """Main functionality for count task"""
 
     i = 0
     input_file, input_annotations = selection
     annotation = Annotation(input_annotations)
-    result = Variant(input_file, annotation)
+    result = Variant(input_file, annotation, skip_files)
 
     if group_by is None:
         for _ in result.read(where=where):
@@ -42,7 +42,7 @@ def _count_task(selection: [str, str], group_by: str, where: str) -> Tuple[int,
 
 
 def count(base_path: str, annotation_path: str or None, group_by: str = None, where: str = None,
-          cores: int = cpu_count(), quite: bool = False) -> Tuple[int, Union[None, dict]]:
+          cores: int = cpu_count(), quite: bool = False, skip_files: bool = False) -> Tuple[int, Union[None, dict]]:
     """Print on the stdout the count result.
 
     It'll parse the input files with its proper annotation schema, and it'll show the count result on the stdout.
@@ -62,6 +62,8 @@ def count(base_path: str, annotation_path: str or None, group_by: str = None, wh
         Discard progress bar.
     cores : int
         Number of cores to parallelize the task.
+    skip_files : bool
+        Skip unreadable files and directories.
 
     Returns
     ----------
@@ -76,7 +78,7 @@ def count(base_path: str, annotation_path: str or None, group_by: str = None, wh
 
     with Pool(cores) as pool:
         groups = {}
-        task = partial(_count_task, group_by=group_by, where=where)
+        task = partial(_count_task, group_by=group_by, where=where, skip_files=skip_files)
         map_method = pool.imap_unordered if len(selection) > 1 else map
         total = 0
         for c, g in tqdm(map_method(task, selection),

openvariant/tasks/groupby.py

@@ -49,7 +49,7 @@ def _group(base_path: str, annotation_path: str or None, key_by: str) -> List[Tu
     return results_by_groups
 
 
-def _group_by_task(selection, where=None, key_by=None, script='', header=False) -> Tuple[str, List, bool]:
+def _group_by_task(selection, where=None, key_by=None, script='', header=False, skip_files=False) -> Tuple[str, List, bool]:
     """Main functionality for group by task"""
     group_key, group_values = selection
 
@@ -59,7 +59,7 @@ def _group_by_task(selection, where=None, key_by=None, script='', header=False)
             for value in group_values:
                 input_file = value[0]
                 annotation = Annotation(value[1])
-                result = Variant(input_file, annotation)
+                result = Variant(input_file, annotation, skip_files)
                 columns = result.annotation.columns if len(result.annotation.columns) != 0 else result.header
 
                 if header:
@@ -83,7 +83,7 @@ def _group_by_task(selection, where=None, key_by=None, script='', header=False)
             for value in group_values:
                 input_file = value[0]
                 annotation = Annotation(value[1])
-                result = Variant(input_file, annotation)
+                result = Variant(input_file, annotation, skip_files)
                 columns = result.annotation.columns if len(result.annotation.columns) != 0 else result.header
 
                 if header:
@@ -116,7 +116,7 @@ def _group_by_task(selection, where=None, key_by=None, script='', header=False)
 
 
 def group_by(base_path: str, annotation_path: str or None, script: str or None, key_by: str, where: str or None = None,
-             cores=cpu_count(), quite=False, header: bool = False) -> Generator[Tuple[str, List, bool], None, None]:
+             cores=cpu_count(), quite=False, header: bool = False, skip_files: bool = False) -> Generator[Tuple[str, List, bool], None, None]:
     """Print on the stdout the group by result.
 
     It'll parse the input files with its proper annotation schema, and it'll show the parsed result separated for each
@@ -141,6 +141,9 @@ def group_by(base_path: str, annotation_path: str or None, script: str or None,
         Number of cores to parallelize the task.
     header : bool
         Number of cores to parallelize the task.
+    skip_files : bool
+        Skip unreadable files and directories.
+
     Returns
     ----------
     int
@@ -150,7 +153,7 @@ def group_by(base_path: str, annotation_path: str or None, script: str or None,
     """
     selection = _group(base_path, annotation_path, key_by)
     with Pool(cores) as pool:
-        task = partial(_group_by_task, where=where, key_by=key_by, script=script, header=header)
+        task = partial(_group_by_task, where=where, key_by=key_by, script=script, header=header, skip_files=skip_files)
         map_method = map if cores == 1 or len(selection) <= 1 else pool.imap_unordered
 
         for group_key, group_result, command in tqdm(