Merge pull request #37 from elixir-europe/update_isa_json_based_on_response

Update isa json based on response

Author: kdp-cloud

.github/workflows/test-mars.yml

@@ -42,3 +42,7 @@ jobs:
       - name: Linting
         run: ruff check mars_lib/
         working-directory: ${{ env.working-directory }}
+
+      - name: Type checking
+        run: mypy --install-types --non-interactive mars_lib/
+        working-directory: ${{ env.working-directory }}

mars-cli/.coveragerc

@@ -1,2 +1,2 @@
 [run]
-omit = mars_lib/__init__.py, mars_lib/submit.py, mars_lib/credential.py
+omit = mars_lib/__init__.py, mars_lib/submit.py, mars_lib/credential.py, mars_lib/models/__init__.py

mars-cli/mars_cli.py

@@ -3,7 +3,7 @@
 import pathlib
 from configparser import ConfigParser
 from mars_lib.target_repo import TargetRepository
-from mars_lib.model import Investigation, IsaJson
+from mars_lib.models.isa_json import Investigation, IsaJson
 from mars_lib.isa_json import load_isa_json
 from logging.handlers import RotatingFileHandler
 import requests
@@ -13,7 +13,7 @@
 
 # Load CLI configuration
 home_dir = (
-    pathlib.Path(os.getenv("MARS_SETTINGS_DIR"))
+    pathlib.Path(str(os.getenv("MARS_SETTINGS_DIR")))
     if os.getenv("MARS_SETTINGS_DIR")
     else pathlib.Path.home()
 )

mars-cli/mars_lib/authentication.py

@@ -1,21 +1,22 @@
+from typing import Optional
 import requests
 import json
 
 
 def get_webin_auth_token(
-    credentials_dict,
-    header={"Content-Type": "application/json"},
-    auth_base_url="https://wwwdev.ebi.ac.uk/ena/dev/submit/webin/auth/token",
-    token_expiration_time=1,
-):
+    credentials_dict: dict[str, str],
+    header: dict[str, str] = {"Content-Type": "application/json"},
+    auth_base_url: str = "https://wwwdev.ebi.ac.uk/ena/dev/submit/webin/auth/token",
+    token_expiration_time: int = 1,
+) -> Optional[str]:
     """
     Obtain Webin authentication token.
 
     Args:
     credentials_dict (dict): The password dictionary for authentication.
     header (dict): The header information.
     auth_base_url (str): The base URL for authentication.
-    token_expiration_time(int): Toke expiration time in hours.
+    token_expiration_time(int): Token expiration time in hours.
 
     Returns:
     str: The obtained token.

mars-cli/mars_lib/biosamples_external_references.py

@@ -6,7 +6,7 @@
 import os
 from jsonschema import validate
 from jsonschema.exceptions import ValidationError, SchemaError
-from typing import Union
+from typing import Union, Any, Optional, List
 
 # -- #
 # Hardcoded values
@@ -21,7 +21,7 @@
 # -- #
 # Code blocks
 # -- #
-def load_json_file(file):
+def load_json_file(file: str) -> Any:
     """
     Function to load a JSON file as a dictionary.
     Args:
@@ -46,7 +46,7 @@ def load_json_file(file):
         )
 
 
-def handle_input_dict(input):
+def handle_input_dict(input: dict[str, str]) -> Optional[dict[str, str]]:
     """
     Function to handle the input: assert that it's either a dictionary or
         the filepath to an existing file containing the dictionary
@@ -73,7 +73,7 @@ def handle_input_dict(input):
             raise ValueError(f"The file '{input}' is not a valid JSON file.")
 
 
-def get_header(token):
+def get_header(token: str) -> dict[str, str]:
     """
     Obtain the header using a token.
 
@@ -90,7 +90,7 @@ def get_header(token):
     }
 
 
-def validate_bs_accession(accession_str):
+def validate_bs_accession(accession_str: str) -> None:
     """
     Validates that the given accession string conforms to the specified regex format.
         See: https://registry.identifiers.org/registry/biosample
@@ -108,8 +108,8 @@ def validate_bs_accession(accession_str):
 
 
 def validate_json_against_schema(
-    json_doc: Union[dict, str], json_schema: Union[dict, str]
-):
+    json_doc: Union[dict[str, List[str]], str], json_schema: Union[dict[str, str], str]
+) -> Optional[bool]:
     """
     Validates a JSON document against a given JSON Schema.
 
@@ -150,7 +150,7 @@ class BiosamplesRecord:
     production: boolean indicating environment mode
     """
 
-    def __init__(self, bs_accession):
+    def __init__(self, bs_accession: str) -> None:
         """
         Initialize the BiosamplesRecord with provided arguments.
 
@@ -159,16 +159,19 @@ def __init__(self, bs_accession):
         """
         validate_bs_accession(bs_accession)
         self.bs_accession = bs_accession
+        self.biosamples_credentials: Optional[dict[str, str]] = None
+        self.biosamples_externalReferences: List[str] = []
+        self.production: bool = False
 
-    def display(self):
+    def display(self) -> None:
         """
         Display the attributes for demonstration purposes.
         """
         print("Biosamples Credentials:", self.biosamples_credentials)
         print("Biosamples External References:", self.biosamples_externalReferences)
         print("Production Mode:", self.production)
 
-    def fetch_bs_json(self, biosamples_endpoint):
+    def fetch_bs_json(self, biosamples_endpoint: str) -> Optional[dict[str, str]]:
         """
         Fetches the BioSample's record (JSON) of the accession.
 
@@ -206,47 +209,49 @@ def fetch_bs_json(self, biosamples_endpoint):
         self.bs_json = response_json
         return self.bs_json
 
-    def load_bs_json(self, bs_json_file: str = None, bs_json: dict = None):
+    def load_bs_json(
+        self, bs_json: Union[str, dict[str, str]]
+    ) -> Optional[dict[str, str]]:
         """
         Loads a given JSON, or the file containing it, as the BioSample's record (JSON) for this instance.
             It is an alternative to fetching it directly from BioSample.
 
         Args:
-            bs_json_file (str): The file containing the Biosamples JSON metadata of the accession
-            bs_json (dict): The already loaded Biosamples JSON metadata of the accession
+            bs_json Union[str, dict]: The already Biosamples JSON metadata of the accession either path to file or dictionary.
         """
-        if bs_json:
-            if isinstance(bs_json, dict):
-                self.bs_json = bs_json
-                return self.bs_json
-            else:
-                raise TypeError(
-                    f"Given 'bs_json' is of type '{type(bs_json)}' instead of type 'dict'."
-                )
-        elif bs_json_file:
-            bs_json = load_json_file(bs_json_file)
+        if isinstance(bs_json, dict):
             self.bs_json = bs_json
             return self.bs_json
+        elif isinstance(bs_json, str):
+            bs_json_data = load_json_file(bs_json)
+            self.bs_json = bs_json_data
+            return self.bs_json
         else:
             raise ValueError(
                 "Neither the file containing the Biosamples JSON nor the Biosamples JSON itself were given to load it into the instance."
             )
 
-    def pop_links(self):
+    def pop_links(self) -> dict[str, str]:
         """
         Removes "_links" array (which is added automatically after updating the biosamples on the BioSample's side).
         """
 
-        if "_links" not in self.bs_json:
-            return self.bs_json
+        if "_links" in self.bs_json:
+            self.bs_json.pop("_links")
 
-        self.bs_json.pop("_links")
         return self.bs_json
 
-    def extend_externalReferences(self, new_ext_refs_list):
+    def extend_externalReferences(
+        self, new_ext_refs_list: List[dict[str, str]]
+    ) -> dict[str, str]:
         """Extends the JSON of the BioSample's record with new externalReferences"""
         if not self.bs_json:
-            self.fetch_bs_json()
+            endpoint = (
+                biosamples_endpoints["prod"]
+                if self.production
+                else biosamples_endpoints["dev"]
+            )
+            self.fetch_bs_json(endpoint)
         self.pop_links()
 
         if "externalReferences" not in self.bs_json:
@@ -265,7 +270,9 @@ def extend_externalReferences(self, new_ext_refs_list):
         self.bs_json["externalReferences"] = ext_refs_list
         return self.bs_json
 
-    def update_remote_record(self, header, webin_auth="?authProvider=WEBIN"):
+    def update_remote_record(
+        self, header: dict[str, str], webin_auth: str = "?authProvider=WEBIN"
+    ) -> Optional[str]:
         """
         Updates the remote record of the BioSample's accession with the current sample JSON.
 

mars-cli/mars_lib/credential.py

@@ -52,27 +52,31 @@
 
 
 class CredentialManager:
-    def __init__(self, service_name):
+    def __init__(self, service_name: str) -> None:
         self.service_name = service_name
 
-    def get_credential_env(self, username):
+    def get_credential_env(self, username: str) -> str:
         """
         Retrieves a credential from environment variables.
 
         :param username: The environment variable username.
         :return: The value of the environment variable or None if not found.
         """
-        return os.getenv(username)
+        result = os.getenv(username)
+        if result is None:
+            raise ValueError(f"Environment variable '{username}' not found.")
 
-    def prompt_for_password(self):
+        return result
+
+    def prompt_for_password(self) -> str:
         """
         Securely prompts the user to enter a password in the console.
 
         :return: The password entered by the user.
         """
         return getpass.getpass(prompt="Enter your password: ")
 
-    def set_password_keyring(self, username, password):
+    def set_password_keyring(self, username: str, password: str) -> None:
         """
         Stores a password in the keyring under the given username.
 
@@ -81,16 +85,19 @@ def set_password_keyring(self, username, password):
         """
         keyring.set_password(self.service_name, username, password)
 
-    def get_password_keyring(self, username):
+    def get_password_keyring(self, username: str) -> str:
         """
         Retrieves a password from the keyring for the given username.
 
         :param username: The username whose password to retrieve.
         :return: The password or None if not found.
         """
-        return keyring.get_password(self.service_name, username)
+        pwd = keyring.get_password(self.service_name, username)
+        if pwd is None:
+            raise ValueError(f"Password not found for username '{username}'.")
+        return pwd
 
-    def delete_password_keyring(self, username):
+    def delete_password_keyring(self, username: str) -> None:
         """
         Deletes a password from the keyring for the given username.