Merge pull request #1361 from Sage-Bionetworks/feature-fds-366-pylint…

…-utils

Feeture: fds-366, linted utils module

.github/workflows/test.yml

@@ -117,6 +117,10 @@ jobs:
           # ran only on certain files for now
           # add here when checked
           poetry run pylint schematic/visualization/* schematic/configuration/*.py schematic/exceptions.py schematic/help.py schematic/loader.py schematic/version.py
+          # do all utils but schema_utils.py
+          poetry run pylint schematic/utils/cli_utils.py schematic/utils/curie_utils.py schematic/utils/df_utils.py
+          poetry run pylint schematic/utils/general.py schematic/utils/google_api_utils.py schematic/utils/io_utils.py
+          poetry run pylint schematic/utils/validate_rules_utils.py schematic/utils/validate_utils.py schematic/utils/viz_utils.py
 
       #----------------------------------------------
       #              run test suite

schematic/manifest/commands.py

@@ -10,7 +10,7 @@
 from schematic.schemas.data_model_graph import DataModelGraph, DataModelGraphExplorer
 from schematic.manifest.generator import ManifestGenerator
 
-from schematic.utils.cli_utils import log_value_from_config, query_dict, parse_synIDs
+from schematic.utils.cli_utils import log_value_from_config, query_dict, parse_syn_ids
 from schematic.utils.google_api_utils import export_manifest_csv
 from schematic.help import manifest_commands
 
@@ -261,7 +261,7 @@ def create_single_manifest(data_type, output_csv=None, output_xlsx=None):
     "-ps",
     "--project_scope",
     default=None,
-    callback=parse_synIDs,
+    callback=parse_syn_ids,
     help=query_dict(manifest_commands, ("manifest", "migrate", "project_scope")),
 )
 @click.option(

schematic/manifest/generator.py

@@ -1504,7 +1504,7 @@ def export_sheet_to_excel(
         export_manifest_drive_service(
             manifest_url,
             file_path=output_excel_file_path,
-            mimeType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+            mime_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
         )
 
         return output_excel_file_path

schematic/models/commands.py

@@ -14,7 +14,7 @@
 from schematic.utils.cli_utils import (
     log_value_from_config,
     query_dict,
-    parse_synIDs,
+    parse_syn_ids,
     parse_comma_str_to_list,
 )
 from schematic.help import model_commands
@@ -98,7 +98,7 @@ def model(ctx, config):  # use as `schematic model ...`
     "-ps",
     "--project_scope",
     default=None,
-    callback=parse_synIDs,
+    callback=parse_syn_ids,
     help=query_dict(model_commands, ("model", "validate", "project_scope")),
 )
 @click.option(
@@ -214,7 +214,7 @@ def submit_manifest(
     "-ps",
     "--project_scope",
     default=None,
-    callback=parse_synIDs,
+    callback=parse_syn_ids,
     help=query_dict(model_commands, ("model", "validate", "project_scope")),
 )
 @click.option(

schematic/utils/cli_utils.py

@@ -1,17 +1,16 @@
-#!/usr/bin/env python3
+"""CLI utils"""
+
+# pylint: disable=logging-fstring-interpolation
+# pylint: disable=anomalous-backslash-in-string
 
-import inspect
 import logging
 
-from typing import Any, Mapping, Sequence, Union, List
+from typing import Any, Mapping, Sequence, Union, Optional
 from functools import reduce
 import re
 
 logger = logging.getLogger(__name__)
 
-# We are using fstrings in logger methods
-# pylint: disable=logging-fstring-interpolation
-
 
 def query_dict(dictionary: Mapping[Any, Any], keys: Sequence[Any]) -> Union[Any, None]:
     """Access a nested value in a dictionary corresponding
@@ -36,7 +35,7 @@ def extract(dictionary: Any, key: Any) -> Union[Any, None]:
     return reduce(extract, keys, dictionary)
 
 
-def log_value_from_config(arg_name: str, config_value: Any):
+def log_value_from_config(arg_name: str, config_value: Any) -> None:
     """Logs when getting a value from the config
 
     Args:
@@ -48,52 +47,56 @@ def log_value_from_config(arg_name: str, config_value: Any):
     )
 
 
-def parse_synIDs(
-    ctx,
-    param,
-    synIDs,
-) -> List[str]:
-    """Parse and validate a comma separated string of synIDs
+def parse_syn_ids(
+    ctx: Any,  # pylint: disable=unused-argument
+    param: str,  # pylint: disable=unused-argument
+    syn_ids: str,
+) -> Optional[list[str]]:
+    """Parse and validate a comma separated string of synapse ids
 
     Args:
-        ctx:
-            click option context
-        param:
-            click option argument name
-        synIDs:
-            comma separated string of synIDs
-
-    Returns:
-        List of synID strings
+        ctx (Any): click option context
+        param (str): click option argument name
+        syn_ids (str): comma separated string of synapse ids
 
     Raises:
-        ValueError: If the entire string does not match a regex for
+        ValueError:  If the entire string does not match a regex for
             a valid comma separated string of SynIDs
+
+    Returns:
+        Optional[list[str]]:  List of synapse ids
     """
-    if synIDs:
-        project_regex = re.compile("(syn\d+\,?)+")
-        valid = project_regex.fullmatch(synIDs)
+    if not syn_ids:
+        return None
 
-        if valid:
-            synIDs = synIDs.split(",")
+    project_regex = re.compile("(syn\d+\,?)+")
+    valid = project_regex.fullmatch(syn_ids)
 
-            return synIDs
+    if not valid:
+        raise ValueError(
+            f"The provided list of project synID(s): {syn_ids}, is not formatted correctly. "
+            "\nPlease check your list of projects for errors."
+        )
 
-        else:
-            raise ValueError(
-                f"The provided list of project synID(s): {synIDs}, is not formatted correctly. "
-                "\nPlease check your list of projects for errors."
-            )
-    else:
-        return
+    return syn_ids.split(",")
 
 
 def parse_comma_str_to_list(
-    ctx,
-    param,
-    comma_string,
-) -> List[str]:
-    if comma_string:
-        return comma_string.split(",")
-    else:
+    ctx: Any,  # pylint: disable=unused-argument
+    param: str,  # pylint: disable=unused-argument
+    comma_string: str,
+) -> Optional[list[str]]:
+    """Separates a comma separated sting into a list of strings
+
+    Args:
+        ctx (Any): click option context
+        param (str): click option argument name
+        comma_string (str): comma separated string
+
+    Returns:
+        Optional[list[str]]: _description_
+    """
+    if not comma_string:
         return None
+
+    return comma_string.split(",")

schematic/utils/curie_utils.py

@@ -1,20 +1,21 @@
+"""Curie utils"""
+
 import logging
 
 
 logger = logging.getLogger(__name__)
 
 
-def extract_name_from_uri_or_curie(item):
+def extract_name_from_uri_or_curie(item: str) -> str:
     """Extract name from uri or curie"""
     if "http" not in item and len(item.split(":")) == 2:
         return item.split(":")[-1]
-    elif len(item.split("//")[-1].split("/")) > 1:
+    if len(item.split("//")[-1].split("/")) > 1:
         return item.split("//")[-1].split("/")[-1]
-    else:
-        raise ValueError("Error extracting name from URI or Curie.")
+    raise ValueError("Error extracting name from URI or Curie.")
 
 
-def expand_curie_to_uri(curie, context_info):
+def expand_curie_to_uri(curie: str, context_info: dict[str, str]) -> str:
     """Expand curie to uri based on the context given
 
     parmas
@@ -24,17 +25,15 @@ def expand_curie_to_uri(curie, context_info):
     "http://schema.biothings.io/"})
     """
     # as suggested in SchemaOrg standard file, these prefixes don't expand
-    PREFIXES_NOT_EXPAND = ["rdf", "rdfs", "xsd"]
+    prefixes_not_expand = ["rdf", "rdfs", "xsd"]
     # determine if a value is curie
     if len(curie.split(":")) == 2:
         prefix, value = curie.split(":")
-        if prefix in context_info and prefix not in PREFIXES_NOT_EXPAND:
+        if prefix in context_info and prefix not in prefixes_not_expand:
             return context_info[prefix] + value
         # if the input is not curie, return the input unmodified
-        else:
-            return curie
-    else:
         return curie
+    return curie
 
 
 def expand_curies_in_schema(schema):
@@ -44,28 +43,28 @@ def expand_curies_in_schema(schema):
     new_schema = {"@context": context, "@graph": [], "@id": schema["@id"]}
     for record in graph:
         new_record = {}
-        for k, v in record.items():
-            if type(v) == str:
-                new_record[expand_curie_to_uri(k, context)] = expand_curie_to_uri(
-                    v, context
+        for key, value in record.items():
+            if isinstance(value, str):
+                new_record[expand_curie_to_uri(key, context)] = expand_curie_to_uri(
+                    value, context
                 )
-            elif type(v) == list:
-                if type(v[0]) == dict:
-                    new_record[expand_curie_to_uri(k, context)] = []
-                    for _item in v:
-                        new_record[expand_curie_to_uri(k, context)].append(
+            elif isinstance(value, list):
+                if isinstance(value[0], dict):
+                    new_record[expand_curie_to_uri(key, context)] = []
+                    for _item in value:
+                        new_record[expand_curie_to_uri(key, context)].append(
                             {"@id": expand_curie_to_uri(_item["@id"], context)}
                         )
                 else:
-                    new_record[expand_curie_to_uri(k, context)] = [
-                        expand_curie_to_uri(_item, context) for _item in v
+                    new_record[expand_curie_to_uri(key, context)] = [
+                        expand_curie_to_uri(_item, context) for _item in value
                     ]
-            elif type(v) == dict and "@id" in v:
-                new_record[expand_curie_to_uri(k, context)] = {
-                    "@id": expand_curie_to_uri(v["@id"], context)
+            elif isinstance(value, dict) and "@id" in value:
+                new_record[expand_curie_to_uri(key, context)] = {
+                    "@id": expand_curie_to_uri(value["@id"], context)
                 }
-            elif v == None:
-                new_record[expand_curie_to_uri(k, context)] = None
+            elif value is None:
+                new_record[expand_curie_to_uri(key, context)] = None
         new_schema["@graph"].append(new_record)
     return new_schema