Improvement/sdev 5193 mypy type fixes

pori_python/graphkb/match.py

@@ -315,15 +315,15 @@ def equivalent_types(
 
     # Convert rid to displayName if needed
     if looks_like_rid(type1):
-        type1 = conn.get_records_by_id([type1])[0]["displayName"]
+        type1 = str(conn.get_records_by_id([type1])[0].get("displayName", type1))
     if looks_like_rid(type2):
-        type2 = conn.get_records_by_id([type2])[0]["displayName"]
+        type2 = str(conn.get_records_by_id([type2])[0].get("displayName", type2))
 
     # Get type terms from observed variant
-    terms1 = []
+    terms1 = set()
     if strict:
         try:
-            terms1.append(get_term_by_name(conn, type1)["@rid"])
+            terms1.add(get_term_by_name(conn, type1)["@rid"])
         except Exception:
             pass
     else:

pori_python/graphkb/util.py

@@ -98,7 +98,7 @@ def cache_key(request_body) -> str:
 class GraphKBConnection:
     def __init__(
         self,
-        url: str = os.environ.get("GRAPHKB_URL"),
+        url: str = os.environ.get("GRAPHKB_URL", ""),
         username: str = "",
         password: str = "",
         use_global_cache: bool = True,
@@ -143,6 +143,8 @@ def request(self, endpoint: str, method: str = "GET", **kwargs) -> Dict:
         Returns:
             dict: the json response as a python dict
         """
+        if not self.url:
+            raise ValueError("no GraphKBConnection url set - cannot make a login demo")
         url = join_url(self.url, endpoint)
         self.request_count += 1
         connect_timeout = 7
@@ -222,6 +224,8 @@ def login_demo(self) -> None:
         1. get a first token from KeyCloak using username and password; self.login_demo()
         2. get a second token from the GraphKB API using keyCloakToken; self.login()
         """
+        if not self.url:
+            raise ValueError("no GraphKBConnection url set - cannot make a login demo")
         url_parts = urlsplit(self.url)
         base_url = f"{url_parts.scheme}://{url_parts.netloc}"
 
@@ -250,8 +254,11 @@ def login(self, username: str, password: str, pori_demo: bool = False) -> None:
         connect_timeout = 7
         read_timeout = 61
 
-        # KBDEV-1328. Alt. GraphKB login for GSC's PORI online demo
-        if pori_demo or "pori-demo" in self.url:
+        if not self.url:
+            raise ValueError("no GraphKBConnection url set - cannot login")
+        elif pori_demo or "pori-demo" in self.url:
+            # KBDEV-1328. Alt. GraphKB login for GSC's PORI online demo
+            logger.warning("login demo")
             self.login_demo()
 
         # use requests package directly to avoid recursion loop on login failure

pori_python/ipr/annotate.py

@@ -103,7 +103,7 @@ def annotate_expression_variants(
     disease_matches: List[str],
     variants: List[IprExprVariant],
     show_progress: bool = False,
-) -> List[KbMatch]:
+) -> List[Hashabledict]:
     """Annotate expression variants with GraphKB in the IPR alterations format.
 
     Args:
@@ -113,10 +113,10 @@ def annotate_expression_variants(
         show_progress (bool): Progressbar displayed for long runs.
 
     Returns:
-        list of kbMatches records for IPR
+        list of Hashabledict records for IPR
     """
     skipped = 0
-    alterations = []
+    alterations: List[Hashabledict] = []
     problem_genes = set()
     logger.info(f"Starting annotation of {len(variants)} expression category_variants")
     iterfunc = tqdm if show_progress else iter
@@ -133,7 +133,7 @@ def annotate_expression_variants(
             for ipr_row in get_ipr_statements_from_variants(graphkb_conn, matches, disease_matches):
                 ipr_row["variant"] = row["key"]
                 ipr_row["variantType"] = row.get("variantType", "exp")
-                alterations.append(ipr_row)
+                alterations.append(Hashabledict(ipr_row))
         except FeatureNotFoundError as err:
             problem_genes.add(gene)
             logger.debug(f"Unrecognized gene ({gene} {variant}): {err}")
@@ -156,7 +156,7 @@ def annotate_copy_variants(
     disease_matches: List[str],
     variants: List[IprCopyVariant],
     show_progress: bool = False,
-) -> List[KbMatch]:
+) -> List[Hashabledict]:
     """Annotate allowed copy variants with GraphKB in the IPR alterations format.
 
     Args:
@@ -169,7 +169,7 @@ def annotate_copy_variants(
         list of kbMatches records for IPR
     """
     skipped = 0
-    alterations = []
+    alterations: List[Hashabledict] = []
     problem_genes = set()
 
     logger.info(f"Starting annotation of {len(variants)} copy category_variants")
@@ -188,7 +188,7 @@ def annotate_copy_variants(
             for ipr_row in get_ipr_statements_from_variants(graphkb_conn, matches, disease_matches):
                 ipr_row["variant"] = row["key"]
                 ipr_row["variantType"] = row.get("variantType", "cnv")
-                alterations.append(ipr_row)
+                alterations.append(Hashabledict(ipr_row))
         except FeatureNotFoundError as err:
             problem_genes.add(gene)
             logger.debug(f"Unrecognized gene ({gene} {variant}): {err}")
@@ -213,7 +213,7 @@ def annotate_positional_variants(
     variants: Sequence[IprStructuralVariant] | Sequence[Hashabledict],
     disease_matches: List[str],
     show_progress: bool = False,
-) -> List[Hashabledict]:
+) -> Sequence[Hashabledict]:
     """Annotate SNP, INDEL or fusion variant calls with GraphKB and return in IPR match format.
 
     Hashable type is required to turn lists into sets.
@@ -239,10 +239,11 @@ def annotate_positional_variants(
             continue
 
         for var_key in VARIANT_KEYS:
-            variant = row.get(var_key)
+            variant = row.get(var_key, "")
             matches = []
             if not variant or isnull(variant):
                 continue
+            variant = str(variant)
             try:
                 try:
                     matches = gkb_match.match_positional_variant(graphkb_conn, variant)
@@ -277,15 +278,15 @@ def annotate_positional_variants(
             except FeatureNotFoundError as err:
                 logger.debug(f"failed to match positional variants ({variant}): {err}")
                 errors += 1
-                if "gene" in row:
-                    problem_genes.add(row["gene"])
-                elif "gene1" in row and f"({row['gene1']})" in str(err):
-                    problem_genes.add(row["gene1"])
-                elif "gene2" in row and f"({row['gene2']})" in str(err):
-                    problem_genes.add(row["gene2"])
-                elif "gene1" in row and "gene2" in row:
-                    problem_genes.add(row["gene1"])
-                    problem_genes.add(row["gene2"])
+                if row.get("gene"):
+                    problem_genes.add(row["gene"])  # type: ignore
+                elif row.get("gene1") and f"({row['gene1']})" in str(err):  # type: ignore
+                    problem_genes.add(row["gene1"])  # type: ignore
+                elif row.get("gene2") and f"({row['gene2']})" in str(err):  # type: ignore
+                    problem_genes.add(row["gene2"])  # type: ignore
+                elif row.get("gene1") and row.get("gene2"):  # type: ignore
+                    problem_genes.add(row["gene1"])  # type: ignore
+                    problem_genes.add(row["gene2"])  # type: ignore
                 else:
                     raise err
             except HTTPError as err:
@@ -314,7 +315,7 @@ def annotate_signature_variants(
     disease_matches: List[str],
     variants: List[IprSignatureVariant] = [],
     show_progress: bool = False,
-) -> List[KbMatch]:
+) -> List[Hashabledict]:
     """Annotate Signature variants with GraphKB in the IPR alterations format.
 
     Match to corresponding GraphKB Variants, then to linked GraphKB Statements
@@ -433,27 +434,21 @@ def annotate_variants(
     # MATCHING COPY VARIANTS
     logger.info(f"annotating {len(copy_variants)} copy variants")
     gkb_matches.extend(
-        [
-            Hashabledict(copy_var)
-            for copy_var in annotate_copy_variants(
-                graphkb_conn, disease_matches, copy_variants, show_progress=interactive
-            )
-        ]
+        annotate_copy_variants(
+            graphkb_conn, disease_matches, copy_variants, show_progress=interactive
+        )
     )
     logger.debug(f"\tgkb_matches: {len(gkb_matches)}")
 
     # MATCHING EXPRESSION VARIANTS
     logger.info(f"annotating {len(expression_variants)} expression variants")
     gkb_matches.extend(
-        [
-            Hashabledict(exp_var)
-            for exp_var in annotate_expression_variants(
-                graphkb_conn,
-                disease_matches,
-                expression_variants,
-                show_progress=interactive,
-            )
-        ]
+        annotate_expression_variants(
+            graphkb_conn,
+            disease_matches,
+            expression_variants,
+            show_progress=interactive,
+        )
     )
     logger.debug(f"\tgkb_matches: {len(gkb_matches)}")
 

pori_python/ipr/connection.py

@@ -16,7 +16,7 @@ def __init__(
         self,
         username: str,
         password: str,
-        url: str = os.environ.get("IPR_URL"),
+        url: str = os.environ.get("IPR_URL", ""),
     ):
         self.token = None
         self.url = url

pori_python/ipr/inputs.py

@@ -11,7 +11,7 @@
 import re
 from Bio.Data.IUPACData import protein_letters_3to1
 from numpy import nan
-from typing import Any, Callable, Dict, Iterable, List, Set, Tuple, cast
+from typing import Any, Callable, Dict, Iterable, List, Sequence, Set, Tuple, cast
 
 from pori_python.graphkb.match import INPUT_COPY_CATEGORIES, INPUT_EXPRESSION_CATEGORIES
 from pori_python.types import (
@@ -26,8 +26,8 @@
 from .constants import (
     COSMIC_SIGNATURE_VARIANT_TYPE,
     HLA_SIGNATURE_VARIANT_TYPE,
-    MSI_MAPPING,
     HRD_MAPPING,
+    MSI_MAPPING,
     TMB_SIGNATURE,
     TMB_SIGNATURE_VARIANT_TYPE,
 )
@@ -248,20 +248,23 @@ def row_key(row: Dict) -> Tuple[str, ...]:
                 row["cnvState"] = display_name_mapping[kb_cat]
         row["variant"] = kb_cat
         row["variantType"] = "cnv"
-        chrband = row.get("chromosomeBand", False)
-        chrom = row.pop("chromosome", False)
-        if not chrom:
-            chrom = row.pop("chr", False)
-        # remove chr if it was not used for chrom
-        row.pop("chr", False)
-        if chrom:
+
+        # Find chromosome and remove chromosome values
+        chrom = ""
+        if "chromosome" in row:
+            chrom = str(row.pop("chromosome", "")) or chrom  # type: ignore
+        if "chr" in row:
+            chrom = str(row.pop("chr", "")) or chrom  # type: ignore
+
+        # Include chromosome in chromosomeBand
+        chrband = row.get("chromosomeBand", "")
+
+        if chrom and chrband:
             # check that chr isn't already in the chrband;
             # this regex from https://vrs.ga4gh.org/en/1.2/terms_and_model.html#id25
-            if chrband and (re.match(r"^cen|[pq](ter|([1-9][0-9]*(\.[1-9][0-9]*)?))$", chrband)):
-                if isinstance(chrom, int):
-                    chrom = str(chrom)
+            if re.match(r"^cen|[pq](ter|([1-9][0-9]*(\.[1-9][0-9]*)?))$", chrband):
                 chrom = chrom.strip("chr")
-                row["chromosomeBand"] = chrom + row["chromosomeBand"]
+                row["chromosomeBand"] = chrom + chrband
 
     return ret_list
 
@@ -441,7 +444,7 @@ def row_key(row: Dict) -> Tuple[str, ...]:
     return result
 
 
-def preprocess_cosmic(rows: Iterable[Dict]) -> Iterable[Dict]:
+def preprocess_cosmic(rows: Iterable[Dict]) -> Sequence[Dict]:
     """
     Process cosmic inputs into preformatted signature inputs
     Note: Cosmic and dMMR already evaluated against thresholds in gsc_report
@@ -456,7 +459,7 @@ def preprocess_cosmic(rows: Iterable[Dict]) -> Iterable[Dict]:
     ]
 
 
-def preprocess_hla(rows: Iterable[Dict]) -> Iterable[Dict]:
+def preprocess_hla(rows: Iterable[Dict]) -> Sequence[Dict]:
     """
     Process hla inputs into preformatted signature inputs
     """
@@ -480,7 +483,7 @@ def preprocess_hla(rows: Iterable[Dict]) -> Iterable[Dict]:
 
 def preprocess_tmb(
     tmb_high: float, tmburMutationBurden: Dict = {}, genomeTmb: float | str = ""
-) -> Iterable[Dict]:
+) -> Sequence[Dict]:
     """
     Process tumour mutation burden (tmb) input(s) into preformatted signature input.
     Get compared to threshold; signature CategoryVariant created only if threshold met.
@@ -530,7 +533,7 @@ def preprocess_tmb(
     return []
 
 
-def preprocess_msi(msi: Any) -> Iterable[Dict]:
+def preprocess_msi(msi: Any) -> Sequence[Dict]:
     """
     Process micro-satellite input into preformatted signature input.
     Both msi & mss gets mapped to corresponding GraphKB Signature CategoryVariants.
@@ -557,7 +560,7 @@ def preprocess_msi(msi: Any) -> Iterable[Dict]:
     return []
 
 
-def preprocess_hrd(hrd: Any) -> Iterable[Dict]:
+def preprocess_hrd(hrd: Any) -> Sequence[Dict]:
     """
     Process hrd input into preformatted signature input.
     HRD gets mapped to corresponding GraphKB Signature CategoryVariants.