aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2023-10-26 07:17:38 +0300
committerFrederick Muriuki Muriithi2023-10-26 07:18:19 +0300
commitc1e488035f2d896a2c4a0e20639c44b381aa2081 (patch)
tree6d30b5e04ef836558dd288e8dc07cfe579cdcd48
parent8c7a9fb61cbcb80edcc7239716b952469bbc6daa (diff)
downloadgenenetwork3-c1e488035f2d896a2c4a0e20639c44b381aa2081.tar.gz
Fix errors caught by mypy.
-rw-r--r--gn3/authentication.py4
-rw-r--r--gn3/case_attributes.py22
-rw-r--r--gn3/commands.py2
-rw-r--r--gn3/computations/gemma.py5
4 files changed, 19 insertions, 14 deletions
diff --git a/gn3/authentication.py b/gn3/authentication.py
index d0b35bc..bb717dd 100644
--- a/gn3/authentication.py
+++ b/gn3/authentication.py
@@ -144,8 +144,8 @@ def get_user_info_by_key(key: str, value: str,
def create_group(conn: Redis, group_name: Optional[str],
- admin_user_uids: List = None,
- member_user_uids: List = None) -> Optional[Dict]:
+ admin_user_uids: Optional[List] = None,
+ member_user_uids: Optional[List] = None) -> Optional[Dict]:
"""Create a group given the group name, members and admins of that group."""
if admin_user_uids is None:
admin_user_uids = []
diff --git a/gn3/case_attributes.py b/gn3/case_attributes.py
index c32e762..d973b8e 100644
--- a/gn3/case_attributes.py
+++ b/gn3/case_attributes.py
@@ -4,6 +4,7 @@ import csv
import json
import uuid
import tempfile
+from typing import Union
from enum import Enum, auto
from pathlib import Path
from functools import reduce
@@ -59,7 +60,9 @@ class CAJSONEncoder(json.JSONEncoder):
return str(obj)
return json.JSONEncoder.default(self, obj)
-def required_access(inbredset_id: int, access_levels: tuple[str, ...]) -> bool:
+def required_access(
+ inbredset_id: int, access_levels: tuple[str, ...]) -> Union[
+ bool, tuple[str, ...]]:
"""Check whether the user has the appropriate access"""
def __species_id__(conn):
with conn.cursor() as cursor:
@@ -330,7 +333,8 @@ def __apply_modifications__(
cattr: new[cattr]
} for cattr in cattrs if new[cattr] != orig[cattr])
- new_rows = reduce(__retrieve_changes__, modifications_diff, tuple())
+ new_rows: tuple[dict, ...] = reduce(
+ __retrieve_changes__, modifications_diff, tuple())
strain_names = tuple({row["Strain"] for row in new_rows})
cursor.execute("SELECT Id AS StrainId, Name AS StrainName FROM Strain "
f"WHERE Name IN ({', '.join(['%s'] * len(strain_names))})",
@@ -454,7 +458,7 @@ def edit_case_attributes(inbredset_id: int) -> Response:
required_access(inbredset_id,
("system:inbredset:edit-case-attribute",))
user = the_token.user
- fieldnames = (["Strain"] + sorted(
+ fieldnames = tuple(["Strain"] + sorted(
attr["Name"] for attr in
__case_attribute_labels_by_inbred_set__(conn, inbredset_id)))
try:
@@ -471,9 +475,9 @@ def edit_case_attributes(inbredset_id: int) -> Response:
conn, inbredset_id),
__inbredset_strains__(conn, inbredset_id)),
__process_edit_data__(
- fieldnames, request.json["edit-data"]))
+ fieldnames, request.json["edit-data"])) # type: ignore[index]
},
- Path(current_app.config.get("TMPDIR"), CATTR_DIFFS_DIR))
+ Path(current_app.config["TMPDIR"], CATTR_DIFFS_DIR))
except NoDiffError as _nde:
msg = "There were no changes to make from submitted data."
response = jsonify({
@@ -502,11 +506,11 @@ def edit_case_attributes(inbredset_id: int) -> Response:
@caseattr.route("/<int:inbredset_id>/diff/list", methods=["GET"])
def list_diffs(inbredset_id: int) -> Response:
"""List any changes that have not been approved/rejected."""
- Path(current_app.config.get("TMPDIR"), CATTR_DIFFS_DIR).mkdir(
+ Path(current_app.config["TMPDIR"], CATTR_DIFFS_DIR).mkdir(
parents=True, exist_ok=True)
def __generate_diff_files__(diffs):
- diff_dir = Path(current_app.config.get("TMPDIR"), CATTR_DIFFS_DIR)
+ diff_dir = Path(current_app.config["TMPDIR"], CATTR_DIFFS_DIR)
review_files = set(afile.name for afile in diff_dir.iterdir()
if ("-rejected" not in afile.name
and "-approved" not in afile.name))
@@ -553,7 +557,7 @@ def list_diffs(inbredset_id: int) -> Response:
@caseattr.route("/approve/<path:filename>", methods=["POST"])
def approve_case_attributes_diff(filename: str) -> Response:
"""Approve the changes to the case attributes in the diff."""
- diff_dir = Path(current_app.config.get("TMPDIR"), CATTR_DIFFS_DIR)
+ diff_dir = Path(current_app.config["TMPDIR"], CATTR_DIFFS_DIR)
diff_filename = Path(diff_dir, filename)
the_diff = __load_diff__(diff_filename)
with database_connection(current_app.config["SQL_URI"]) as conn:
@@ -566,7 +570,7 @@ def approve_case_attributes_diff(filename: str) -> Response:
@caseattr.route("/reject/<path:filename>", methods=["POST"])
def reject_case_attributes_diff(filename: str) -> Response:
"""Reject the changes to the case attributes in the diff."""
- diff_dir = Path(current_app.config.get("TMPDIR"), CATTR_DIFFS_DIR)
+ diff_dir = Path(current_app.config["TMPDIR"], CATTR_DIFFS_DIR)
diff_filename = Path(diff_dir, filename)
the_diff = __load_diff__(diff_filename)
with database_connection(current_app.config["SQL_URI"]) as conn:
diff --git a/gn3/commands.py b/gn3/commands.py
index a90e895..b873d4d 100644
--- a/gn3/commands.py
+++ b/gn3/commands.py
@@ -148,7 +148,7 @@ def run_sample_corr_cmd(method, this_trait_data, target_dataset_data):
return correlation_results
-def run_cmd(cmd: str, success_codes: Tuple = (0,), env: str = None) -> Dict:
+def run_cmd(cmd: str, success_codes: Tuple = (0,), env: Optional[str] = None) -> Dict:
"""Run CMD and return the CMD's status code and output as a dict"""
parsed_cmd = json.loads(cmd)
parsed_env = (json.loads(env) if env is not None else None)
diff --git a/gn3/computations/gemma.py b/gn3/computations/gemma.py
index 8036a7b..7f09b0a 100644
--- a/gn3/computations/gemma.py
+++ b/gn3/computations/gemma.py
@@ -5,6 +5,7 @@ from base64 import b64encode
from hashlib import md5
from typing import Dict
from typing import List
+from typing import Optional
from typing import ValuesView
from gn3.commands import compose_gemma_cmd
from gn3.fs_helpers import get_hash_of_files
@@ -53,8 +54,8 @@ def generate_gemma_cmd(gemma_cmd: str,
output_dir: str,
token: str,
gemma_kwargs: Dict,
- gemma_wrapper_kwargs: Dict = None,
- chromosomes: str = None) -> Dict:
+ gemma_wrapper_kwargs: Optional[Dict] = None,
+ chromosomes: Optional[str] = None) -> Dict:
"""Compute k values"""
_hash = get_hash_of_files(
[v for k, v in gemma_kwargs.items() if k in ["g", "p", "a", "c"]])