WIP typing

Author: eboileau

server/pyproject.toml

@@ -77,13 +77,22 @@ where = ["src"]
 
 [tool.mypy]
 plugins = [
-  "sqlalchemy.ext.mypy.plugin"
+  "sqlalchemy.ext.mypy.plugin",
+  "pydantic.mypy"
 ]
 show_error_codes = "True"
+# follow_imports = "silent"
+# warn_redundant_casts = true
+# warn_unused_ignores = true
+# disallow_any_generics = true
+# check_untyped_defs = true
+# no_implicit_reexport = true
 #strict = "True"
-exclude = [
-    "scimodom.utils.models.py"
-]
+
+# [tool.pydantic-mypy]
+# init_forbid_extra = true
+# init_typed = true
+# warn_required_dynamic_aliases = true
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

server/src/scimodom/api/dataset.py

@@ -1,6 +1,6 @@
 import logging
 from dataclasses import dataclass
-from typing import Iterable
+from typing import Generator, Iterable, Sequence
 
 from flask import Blueprint
 from flask_cors import cross_origin
@@ -110,7 +110,7 @@ class _CompareContext:
     class Ctx:
         bedtools_service: BedToolsService
         a_records: Iterable[ComparisonRecord]
-        b_records_list: list[Iterable[ComparisonRecord]]
+        b_records_list: Sequence[Iterable[ComparisonRecord]]
         is_strand: bool
 
     def __init__(self):
@@ -170,7 +170,9 @@ def __enter__(self) -> Ctx:
             is_strand=self._is_strand,
         )
 
-    def _get_comparison_records_from_db(self, dataset_ids):
+    def _get_comparison_records_from_db(
+        self, dataset_ids
+    ) -> Generator[ComparisonRecord, None, None]:
         for dataset_id in dataset_ids:
             for data in self._data_service.get_by_dataset(dataset_id):
                 yield ComparisonRecord(
@@ -185,8 +187,10 @@ def _get_comparison_records_from_db(self, dataset_ids):
                     frequency=data.frequency,
                 )
 
-    def _get_comparison_records_from_file(self):
-        dummy_values = {"eufid": "upload      "}
+    def _get_comparison_records_from_file(
+        self,
+    ) -> Generator[ComparisonRecord, None, None]:
+        dummy_values: dict[str, str | int] = {"eufid": "upload      "}
         if self._is_euf:
             importer = EufImporter(
                 stream=self._tmp_file_handle, source=self._upload_name

server/src/scimodom/api/helpers.py

@@ -169,13 +169,18 @@ def validate_request_size(max_size) -> None:
         raise ClientResponseException(413, f"File too large (max. {max_size} bytes)")
 
 
-def get_valid_taxa_id(raw: str) -> int:
-    utilities_service = get_utilities_service()
-    taxa_ids = [d["id"] for d in utilities_service.get_taxa()]
+def get_valid_taxa_id() -> int:
+    taxa_id = request.args.get("taxaId", type=int)
+    if taxa_id is None:
+        raise ClientResponseException(400, "Invalid Taxa ID")
+    _validate_taxa_id(taxa_id)
+    return taxa_id
+
+
+def get_valid_taxa_id_from_string(raw: str) -> int:
     try:
         taxa_id = int(raw)
-        if taxa_id not in taxa_ids:
-            raise ClientResponseException(404, "Unrecognized Taxa ID")
+        _validate_taxa_id(taxa_id)
         return taxa_id
     except ValueError:
         raise ClientResponseException(400, "Invalid Taxa ID")
@@ -197,8 +202,14 @@ def get_valid_targets_type(raw: str) -> TargetsFileType:
 
 def get_valid_coords(taxa_id: int, context: int = 0) -> tuple[str, int, int, Strand]:
     chrom = request.args.get("chrom", type=str)
+    if chrom is None:
+        raise ClientResponseException(400, "Invalid chrom")
     start = request.args.get("start", type=int)
+    if start is None:
+        raise ClientResponseException(400, "Invalid start")
     end = request.args.get("end", type=int)
+    if end is None:
+        raise ClientResponseException(400, "Invalid end")
     strand = request.args.get("strand", default=".", type=str)
 
     assembly_service = get_assembly_service()
@@ -285,3 +296,10 @@ def _is_valid_identifier(identifier, length):
     elif len(identifier) != length:
         return False
     return True
+
+
+def _validate_taxa_id(taxa_id: int) -> None:
+    utilities_service = get_utilities_service()
+    taxa_ids = [d["id"] for d in utilities_service.get_taxa()]
+    if taxa_id not in taxa_ids:
+        raise ClientResponseException(404, "Unrecognized Taxa ID")

server/src/scimodom/api/modification.py

@@ -100,7 +100,7 @@ def get_modifications_as_csv(by_gene):
 def get_modification_sitewise():
     """Get information related to a modification site."""
     try:
-        taxa_id = get_valid_taxa_id(request.args.get("taxaId"))
+        taxa_id = get_valid_taxa_id()
         chrom, start, end, _ = get_valid_coords(taxa_id)
     except ClientResponseException as e:
         return e.response_tuple
@@ -118,7 +118,7 @@ def get_modification_sitewise():
 def get_genomic_sequence_context(context):
     """Get sequence context for a modification site."""
     try:
-        taxa_id = get_valid_taxa_id(request.args.get("taxaId"))
+        taxa_id = get_valid_taxa_id()
         coords = get_valid_coords(taxa_id, context=int(context))
     except ClientResponseException as e:
         return e.response_tuple
@@ -168,7 +168,7 @@ class Ctx:
     def __init__(self, target_type: str):
         self._target_type = get_valid_targets_type(target_type)
         self._is_strand = True
-        self._taxa_id = get_valid_taxa_id(request.args.get("taxaId"))
+        self._taxa_id = get_valid_taxa_id()
         self._coords = get_valid_coords(self._taxa_id)
 
     def __enter__(self) -> Ctx:
@@ -222,7 +222,7 @@ def _get_modifications_for_request(by_gene):
         gene_or_chrom = _get_gene_or_chrom_required()
         return modification_service.get_modifications_by_gene(
             annotation_source=_get_annotation_source(),
-            taxa_id=get_valid_taxa_id(request.args.get("taxaId")),
+            taxa_id=get_valid_taxa_id(),
             gene_filter=gene_or_chrom.gene_filter,
             chrom=gene_or_chrom.chrom_filter,
             chrom_start=gene_or_chrom.chrom_start_filter,
@@ -237,7 +237,7 @@ def _get_modifications_for_request(by_gene):
             modification_id=get_non_negative_int("modification"),
             organism_id=get_non_negative_int("organism"),
             technology_ids=_get_technology_ids(),
-            taxa_id=get_valid_taxa_id(request.args.get("taxaId")),
+            taxa_id=get_valid_taxa_id(),
             gene_filter=_get_gene_filters(),
             chrom=request.args.get("chrom", type=str),
             chrom_start=get_optional_non_negative_int("chromStart"),

server/src/scimodom/api/utilities.py

@@ -4,7 +4,7 @@
 
 from scimodom.api.helpers import (
     get_valid_logo,
-    get_valid_taxa_id,
+    get_valid_taxa_id_from_string,
     ClientResponseException,
     validate_rna_type,
 )
@@ -93,7 +93,7 @@ def get_features(rna_type):
 def get_chroms(taxa_id: str):
     assembly_service = get_assembly_service()
     try:
-        taxa_id_as_int = get_valid_taxa_id(taxa_id)
+        taxa_id_as_int = get_valid_taxa_id_from_string(taxa_id)
         return assembly_service.get_chroms(taxa_id_as_int)
     except ClientResponseException as e:
         return e.response_tuple
@@ -108,7 +108,7 @@ def get_chroms(taxa_id: str):
 def get_assemblies(taxa_id):
     utilities_service = get_utilities_service()
     try:
-        taxa_id_as_int = get_valid_taxa_id(taxa_id)
+        taxa_id_as_int = get_valid_taxa_id_from_string(taxa_id)
         return utilities_service.get_assemblies(taxa_id_as_int)
     except ClientResponseException as e:
         return e.response_tuple

server/src/scimodom/cli/project.py

@@ -162,10 +162,14 @@ def create_project_template(
         sources = []
         for dv, pv in zip(doi, pmid):
             if dv == "null":
-                dv = None
+                dv_dto = None
+            else:
+                dv_dto = dv
             if pv == 0:
-                pv = None
-            source = ProjectSourceDto(doi=dv, pmid=pv)
+                pv_dto = None
+            else:
+                pv_dto = pv
+            source = ProjectSourceDto(doi=dv_dto, pmid=pv_dto)
             sources.append(source)
 
         project_template = ProjectTemplate(

server/src/scimodom/services/annotation/gtrnadb.py

@@ -82,7 +82,8 @@ def create_annotation(self, taxa_id: int, **kwargs) -> None:
         fasta_file = annotation_paths["fa"].annotation_file
         # TODO
         # seqids = self._assembly_service.get_seqids(annotation.taxa_id)
-        seqids = []
+        # ?
+        # seqids = []
 
         logger.info(
             f"Setting up GtRNAdb {annotation.release} for {annotation.taxa_id}..."

server/src/scimodom/services/assembly.py

@@ -2,7 +2,7 @@
 import logging
 from functools import cache
 from posixpath import join as urljoin
-from typing import Any
+from typing import Any, Sequence
 
 from sqlalchemy import select, func
 from sqlalchemy.exc import NoResultFound
@@ -86,7 +86,7 @@ def get_assembly_by_id(self, assembly_id: int) -> Assembly:
         except NoResultFound:
             raise AssemblyNotFoundError(f"No such assembly with ID: {assembly_id}.")
 
-    def get_assemblies_by_taxa(self, taxa_id: int) -> list[Assembly]:
+    def get_assemblies_by_taxa(self, taxa_id: int) -> Sequence[Assembly]:
         """Retrieve all assemblies for a given organism.
 
         :param taxa_id: Taxonomy ID

server/src/scimodom/services/exporter.py

@@ -19,7 +19,7 @@
     Annotation,
     AnnotationVersion,
 )
-from scimodom.utils.specs.euf import EUF
+from scimodom.utils.specs.euf import EUF_VERSION
 
 logger = logging.getLogger(__name__)
 
@@ -34,7 +34,7 @@ class NoSuchDataset(Exception):
 
 class Exporter:
     BAD_FILE_NAME_CHARACTERS_REGEXP = re.compile(r"[^a-zA-Z0-9(),._-]")
-    VERSION = EUF["versions"][-1]
+    VERSION = EUF_VERSION
 
     def __init__(self, session: Session):
         self._session = session

server/src/scimodom/services/file.py

@@ -113,7 +113,7 @@ def open_file_for_reading(path: str | Path) -> TextIO:
         return open(path, "r")
 
     @staticmethod
-    def count_lines(path: str | Path):
+    def count_lines(path: str | Path) -> int:
         count = 0
         with open(path) as fp:
             while True:
@@ -124,7 +124,7 @@ def count_lines(path: str | Path):
 
     # Import
 
-    def check_import_file(self, name: str):
+    def check_import_file(self, name: str) -> bool:
         return Path(self._import_path, name).is_file()
 
     def open_import_file(self, name: str):
@@ -182,7 +182,7 @@ def get_gene_cache(self, selection_ids: Iterable[int]) -> list[str]:
         :returns: The genes of the selection
         :rtype: list[str]
         """
-        result = set()
+        result: set = set()
         for selection_id in selection_ids:
             path = Path(self._get_gene_cache_dir(), str(selection_id))
             with open(path) as fh:
@@ -237,7 +237,9 @@ def open_sunburst_cache(self, name: str) -> TextIO:
         file_path = Path(self._get_sunburst_cache_dir(), f"{name}.json")
         return open(file_path)
 
-    def update_sunburst_cache(self, name: str, generator: Generator[str, None, None]):
+    def update_sunburst_cache(
+        self, name: str, generator: Generator[str, None, None]
+    ) -> None:
         final_file_path = Path(self._get_sunburst_cache_dir(), f"{name}.json")
         try:
             temporary_file_path = None
@@ -335,10 +337,10 @@ def delete_project_request_file(self, request_uuid) -> None:
     def _get_project_metadata_dir(self) -> Path:
         return Path(self._data_path, self.METADATA_DEST)
 
-    def _get_project_request_file_path(self, request_uuid):
+    def _get_project_request_file_path(self, request_uuid) -> Path:
         return Path(self._get_project_request_dir(), f"{request_uuid}.json")
 
-    def _get_project_request_dir(self):
+    def _get_project_request_dir(self) -> Path:
         return Path(self._get_project_metadata_dir(), self.REQUEST_DEST)
 
     # Assembly
@@ -488,7 +490,7 @@ def _get_current_assembly_name_from_taxa_id(self, taxa_id: int) -> str:
 
     # uploaded files
 
-    def upload_tmp_file(self, stream, max_file_size):
+    def upload_tmp_file(self, stream, max_file_size) -> str:
         fp, path = mkstemp(dir=self._upload_path)
         close(fp)
         file_id = basename(path)
@@ -509,7 +511,9 @@ def check_tmp_upload_file_id(self, file_id: str) -> bool:
         path = join(self._upload_path, file_id)
         return isfile(path)
 
-    def _stream_to_file(self, data_stream, path, max_size, overwrite_is_ok=False):
+    def _stream_to_file(
+        self, data_stream, path, max_size, overwrite_is_ok=False
+    ) -> None:
         if exists(path) and not overwrite_is_ok:
             raise Exception(
                 f"FileService._stream_to_file(): Refusing to overwrite existing file: '{path}'!"
@@ -555,7 +559,7 @@ def create_or_update_bam_file(
         name: str,
         data_stream: IO[bytes],
         max_size: Optional[int],
-    ):
+    ) -> None:
         try:
             bam_file = self.get_bam_file(dataset, name)
             self._update_bam_file(bam_file, data_stream, max_size)
@@ -581,7 +585,7 @@ def get_bam_file_list(self, dataset: Dataset) -> List[Dict[str, Any]]:
         ).all()
         return [self._get_bam_file_info(i) for i in items]
 
-    def remove_bam_file(self, bam_file: BamFile):
+    def remove_bam_file(self, bam_file: BamFile) -> None:
         path = self._get_bam_file_path(bam_file)
         try:
             unlink(path)
@@ -591,7 +595,7 @@ def remove_bam_file(self, bam_file: BamFile):
         self._session.commit()
 
     @staticmethod
-    def _handle_upload_error(exception, path):
+    def _handle_upload_error(exception, path) -> None:
         logger.warning(
             f"Failed to create file '{path}': {str(exception)} - discarding file."
         )
@@ -601,10 +605,10 @@ def _handle_upload_error(exception, path):
             logger.warning(f"Failed to delete '{path}': {str(unlink_e)}.")
         raise exception
 
-    def _get_bam_files_parent_dir(self):
+    def _get_bam_files_parent_dir(self) -> Path:
         return Path(self._data_path, self.BAM_DEST)
 
-    def _update_bam_file(self, bam_file, data_stream, max_size):
+    def _update_bam_file(self, bam_file, data_stream, max_size) -> None:
         tmp_path = self._get_bam_file_tmp_path(bam_file)
         path = self._get_bam_file_path(bam_file)
         self._stream_to_file(data_stream, tmp_path, max_size)
@@ -621,18 +625,18 @@ def _update_bam_file(self, bam_file, data_stream, max_size):
             self._session.delete(bam_file)
             raise exc
 
-    def _get_bam_file_tmp_path(self, bam_file):
+    def _get_bam_file_tmp_path(self, bam_file) -> str:
         return join(
             self._get_bam_files_parent_dir().as_posix(),
             f"tmp.{bam_file.storage_file_name}",
         )
 
-    def _get_bam_file_path(self, bam_file):
+    def _get_bam_file_path(self, bam_file) -> str:
         return join(
             self._get_bam_files_parent_dir().as_posix(), bam_file.storage_file_name
         )
 
-    def _create_bam_file(self, dataset, name, data_stream, max_size):
+    def _create_bam_file(self, dataset, name, data_stream, max_size) -> None:
         bam_file = BamFile(
             dataset_id=dataset.id,
             original_file_name=name,
@@ -643,7 +647,7 @@ def _create_bam_file(self, dataset, name, data_stream, max_size):
         self._session.add(bam_file)
         self._session.commit()
 
-    def _get_bam_file_info(self, bam_file):
+    def _get_bam_file_info(self, bam_file) -> dict[str, Any]:
         path = self._get_bam_file_path(bam_file)
         stat_info = stat(path)
         return {