add: type checking

.github/workflows/main.yml

@@ -28,6 +28,8 @@ jobs:
         run: pylint pro_wes/ setup.py
       - name: Check code format with Black
         run: black --check setup.py pro_wes/ tests/
+      - name: Type check with mypy
+        run: mypy setup.py pro_wes/
   test:
     name: Run tests
     runs-on: ubuntu-latest

mypy.ini

@@ -0,0 +1,6 @@
+[mypy]
+ignore_missing_imports = True
+disable_error_code = attr-defined
+disable_error_code = no-redef
+
+# [mypy-pro_wes.*]

pro_wes/app.py

@@ -5,6 +5,7 @@
 from connexion import App
 from flask import current_app
 from foca import Foca
+from typing import Dict
 
 from pro_wes.ga4gh.wes.service_info import ServiceInfo
 from pro_wes.exceptions import NotFound
@@ -32,13 +33,14 @@ def _setup_first_start(app: App) -> None:
         work_dir = Path(current_app.config.foca.custom.post_runs.storage_path.resolve())
         work_dir.mkdir(parents=True, exist_ok=True)
         # set service info
-        service_info = ServiceInfo()
+        service_info: Dict
         try:
-            service_info = service_info.get_service_info()
+            service_info = ServiceInfo().get_service_info()
         except NotFound:
-            service_info.set_service_info(
-                data=current_app.config.foca.custom.service_info.dict()
-            )
+            service_info_data: Dict = current_app.config.foca.custom.service_info.dict()
+            service_info_object = ServiceInfo()
+            service_info_object.set_service_info(data=service_info_data)
+            service_info = service_info_object.get_service_info()  # noqa: F841
 
 
 def run_app(app: App) -> None:

pro_wes/ga4gh/wes/models.py

@@ -38,10 +38,12 @@ class Attachment(BaseModel):
 
     Args:
         filename: Name of the file as indicated in the run request.
+        object: File object.
         path: Path to the file on the app's storage system.
     """
 
     filename: str
+    object: bytes
     path: Path
 
 
@@ -171,7 +173,7 @@ def json_serialized_object_field_valid(  # pylint: disable=no-self-argument
     def workflow_type_and_version_supported(  # pylint: disable=no-self-argument
         cls,
         values: Dict,
-    ) -> str:
+    ) -> Dict:
         """Ensure that workflow type and version are supported by this service
         instance.
 
@@ -338,7 +340,7 @@ class WesEndpoint(BaseModel):
 
     host: str
     base_path: Optional[str] = "/ga4gh/wes/v1"
-    run_id: Optional[str]
+    run_id: Optional[str] = None
 
 
 class DbDocument(BaseModel):

pro_wes/ga4gh/wes/workflow_runs.py

@@ -96,13 +96,22 @@ def run_workflow(
         # write workflow attachments
         self._save_attachments(attachments=document_stored.attachments)
 
+        if document_stored.wes_endpoint is None:
+            raise ValueError("No WES endpoint available.")
+
+        if document_stored.wes_endpoint.base_path is None:
+            raise ValueError("No WES endpoint base path provided.")
+
         # instantiate WES client
         wes_client: WesClient = WesClient(
             host=document_stored.wes_endpoint.host,
             base_path=document_stored.wes_endpoint.base_path,
             token=kwargs.get("jwt", None),
         )
 
+        if document_stored.task_id is None:
+            raise ValueError("No task ID available")
+
         # instantiate database connector
         db_connector = DbDocumentConnector(
             collection=self.db_client,
@@ -136,25 +145,34 @@ def run_workflow(
             if response.status_code == 403:
                 raise Forbidden
             raise InternalServerError
-        document_stored: DbDocument = db_connector.upsert_fields_in_root_object(
+        updated_document_stored: DbDocument = db_connector.upsert_fields_in_root_object(
             root="wes_endpoint",
             run_id=response.run_id,
         )
 
+        if updated_document_stored.wes_endpoint is None:
+            raise ValueError("No WES endpoint available.")
+
         # track workflow progress in background
         task__track_run_progress.apply_async(
             None,
             {
                 "jwt": kwargs.get("jwt", None),
-                "remote_host": document_stored.wes_endpoint.host,
-                "remote_base_path": document_stored.wes_endpoint.base_path,
-                "remote_run_id": document_stored.wes_endpoint.run_id,
+                "remote_host": updated_document_stored.wes_endpoint.host,
+                "remote_base_path": updated_document_stored.wes_endpoint.base_path,
+                "remote_run_id": updated_document_stored.wes_endpoint.run_id,
             },
-            task_id=document_stored.task_id,
+            task_id=updated_document_stored.task_id,
             soft_time_limit=controller_config.timeout_job,
         )
 
-        return {"run_id": document_stored.run_log.run_id}
+        if updated_document_stored.wes_endpoint is None:
+            raise ValueError("No WES endpoint available.")
+
+        if updated_document_stored.wes_endpoint.run_id is None:
+            raise ValueError("WES endpoint does not have run_id.")
+
+        return {"run_id": updated_document_stored.wes_endpoint.run_id}
 
     def list_runs(
         self,
@@ -406,7 +424,7 @@ def _create_run_environment(
             # populate document
             document.run_log.run_id = run_id
             document.task_id = uuid()
-            document.work_dir = str(work_dir)
+            document.work_dir = work_dir.as_posix()
             document.attachments = self._process_attachments(
                 work_dir=work_dir,
             )
@@ -437,18 +455,20 @@ def _process_attachments(self, work_dir: Path) -> List[Attachment]:
         attachments = []
         files = request.files.getlist("workflow_attachment")
         for file in files:
-            attachments.append(
-                Attachment(
-                    filename=file.filename,
-                    object=file.stream,
-                    path=str(
-                        work_dir
+            if file is not None:
+                if file.filename is None:
+                    raise ValueError("File does not have a filename.")
+
+                attachments.append(
+                    Attachment(
+                        filename=file.filename,
+                        object=file.stream.read(),
+                        path=work_dir
                         / self._secure_filename(
                             name=Path(file.filename),
-                        )
-                    ),
+                        ),
+                    )
                 )
-            )
         return attachments
 
     @staticmethod

pro_wes/tasks/track_run_progress.py

@@ -97,7 +97,7 @@ def task__track_run_progress(
     response.pop("request", None)
     document: DbDocument = db_client.upsert_fields_in_root_object(
         root="run_log",
-        **response,
+        **dict(response),
     )
 
     # track workflow run progress
@@ -106,7 +106,13 @@ def task__track_run_progress(
     while not run_state.is_finished:
         sleep(controller_config.polling_wait)
         try:
-            response = wes_client.get_run_status(
+            if document.wes_endpoint is None:
+                raise ValueError("WES run ID not available.")
+
+            if document.wes_endpoint.run_id is None:
+                raise ValueError("WES run ID not available.")
+
+            wes_client.get_run_status(
                 run_id=document.wes_endpoint.run_id,
                 timeout=foca_config.custom.defaults.timeout,
             )
@@ -133,6 +139,9 @@ def task__track_run_progress(
     try:
         # workaround for cwl-WES; add .dict() when cwl-WES response conforms
         # to model
+        if response.run_id is None:
+            raise ValueError("WES run ID not available.")
+
         response = wes_client.get_run(run_id=response.run_id)
     except EngineUnavailable:
         db_client.update_run_state(state=State.SYSTEM_ERROR.value)
@@ -143,7 +152,8 @@ def task__track_run_progress(
     response.pop("request", None)
     document = db_client.upsert_fields_in_root_object(
         root="run_log",
-        **response,
+        **dict(response),
     )
 
     logger.info(f"[{self.request.id}] Processing completed.")
+    return self.request.id

pro_wes/utils/db.py

@@ -88,7 +88,7 @@ def upsert_fields_in_root_object(
         self,
         root: str,
         **kwargs,
-    ) -> Optional[Mapping]:
+    ) -> DbDocument:
         """Insert (or update) fields in(to) the same root object and return
         document.
         """

requirements_dev.txt

@@ -5,3 +5,6 @@ mongomock>=4.1.2,<5
 pylint>=2.15.5,<3
 pytest>=7.2.0,<8
 python-semantic-release>=7.32.2,<8
+mypy
+types-setuptools
+types-requests

setup.py

@@ -16,7 +16,7 @@
 
 setup(
     name="pro-wes",
-    version=__version__,  # noqa: F821  # pylint: disable=undefined-variable
+    version=__version__,  # type: ignore # noqa: E501 F821 # pylint: disable=undefined-variable
     license="Apache License 2.0",
     description="Proxy/gateway GA4GH WES service",
     long_description=LONG_DESCRIPTION,