Merge pull request #459 from xchem/m2ms-1240-achristie

style: Remaining mypy changes (except target loader)

.pre-commit-config.yaml

@@ -49,8 +49,8 @@ repos:
 # MyPy
 #- repo: https://github.com/pre-commit/mirrors-mypy
 #  rev: v1.7.1
-#  - id: mypy
 #  hooks:
+#  - id: mypy
 #    additional_dependencies:
 #    - types-PyYAML
 #    - types-pymysql

viewer/download_structures.py

@@ -13,6 +13,7 @@
 import uuid
 import zipfile
 from pathlib import Path
+from typing import Any, Dict
 
 import pandoc
 from django.conf import settings
@@ -557,7 +558,7 @@ def _create_structures_dict(target, site_obvs, protein_params, other_params):
     """
     logger.info('+ _create_structures_dict')
 
-    zip_contents = copy.deepcopy(zip_template)
+    zip_contents: Dict[str, Any] = copy.deepcopy(zip_template)
 
     logger.info(
         'Expecting %d molecules from %d proteins', site_obvs.count(), site_obvs.count()

viewer/services.py

@@ -200,6 +200,8 @@ def keycloak(func_id, name, url=None, secret=None):
 
     logger.debug("+ keycloak")
     keycloak_realm = os.environ.get(url, None)
+    if not keycloak_realm:
+        return False
     response = requests.get(keycloak_realm)
     logger.debug("keycloak response: %s", response)
     return response.ok

viewer/tasks.py

@@ -2,6 +2,7 @@
 import logging
 import os
 import shutil
+from typing import Any, Dict
 
 from django.db import IntegrityError
 
@@ -178,7 +179,11 @@ def validate_compound_set(task_params):
     logger.info('validate_compound_set() update=%s', update)
 
     validated = True
-    validate_dict = {'molecule_name': [], 'field': [], 'warning_string': []}
+    validate_dict: Dict[str, Any] = {
+        'molecule_name': [],
+        'field': [],
+        'warning_string': [],
+    }
 
     # params = {
     #     'sdf': 'tests/test_data/test_chodera/test_0_2.sdf',

viewer/tests/test_loader.py

@@ -43,13 +43,15 @@ def setUpTestData(cls):
 
     @classmethod
     def tearDownClass(cls):
-        cls.tempdir.cleanup()
+        if cls.tempdir:
+            cls.tempdir.cleanup()
         super().tearDownClass()
 
     def test__process_quat_assembly_positive(self):
         idx = next(iter(self.assemblies))
         data = self.assemblies[idx]
 
+        assert self.target_loader
         result = self.target_loader._process_quat_assembly(  # pylint: disable=protected-access
             QuatAssembly.objects.none(), idx, data
         )  # pylint: disable=protected-access

viewer/views.py

@@ -7,7 +7,7 @@
 import zipfile
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 from wsgiref.util import FileWrapper
 
 import pandas as pd
@@ -258,7 +258,7 @@ def save_pdb_zip(pdb_file):
     zf = zipfile.ZipFile(pdb_file)
     zip_lst = zf.namelist()
     zfile = {}
-    zfile_hashvals = {}
+    zfile_hashvals: Dict[str, str] = {}
     print(zip_lst)
     for filename in zip_lst:
         # only handle pdb files
@@ -338,11 +338,9 @@ def get(self, request):
         # - this can be switched off in settings.py.
         user = self.request.user
         if not user.is_authenticated and settings.AUTHENTICATE_UPLOAD:
-            context = {}
-            context['error_message'] = (
-                'Only authenticated users can upload files'
-                ' - please navigate to landing page and Login'
-            )
+            context: Dict[str, Any] = {
+                'error_message': 'Only authenticated users can upload files - please navigate to landing page and Login'
+            }
             return render(request, 'viewer/upload-cset.html', context)
 
         form = CSetForm()
@@ -397,7 +395,7 @@ def post(self, request):
 
             # If a set is named the ComputedSet cannot be 'Anonymous'
             # and the user has to be the owner.
-            selected_set = None
+            selected_set: Optional[models.ComputedSet] = None
             if update_set and update_set != 'None':
                 computed_set_query = models.ComputedSet.objects.filter(
                     unique_name=update_set
@@ -539,6 +537,7 @@ def post(self, request):
 
             elif choice == 'D':
                 # Delete
+                assert selected_set
                 selected_set.delete()
 
                 request.session[
@@ -1719,13 +1718,15 @@ def create(self, request):
         # Only authenticated users can transfer files to sqonk
         user = self.request.user
         if not user.is_authenticated:
-            content = {'Only authenticated users can transfer files'}
+            content: Dict[str, Any] = {
+                'error': 'Only authenticated users can transfer files'
+            }
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Can't use this method if the Squonk2 agent is not configured
         sq2a_rv = _SQ2A.configured()
         if not sq2a_rv.success:
-            content = {f'The Squonk2 Agent is not configured ({sq2a_rv.msg})'}
+            content = {'error': f'The Squonk2 Agent is not configured ({sq2a_rv.msg})'}
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Collect expected API parameters....
@@ -1753,7 +1754,7 @@ def create(self, request):
         )
         sq2a_rv = _SQ2A.can_send(sq2a_send_params)
         if not sq2a_rv.success:
-            content = {f'You cannot do this ({sq2a_rv.msg})'}
+            content = {'error': f'You cannot do this ({sq2a_rv.msg})'}
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Check the existence of the files that are expected to be transferred
@@ -1898,7 +1899,9 @@ def create(self, request):
         # Only authenticated users can transfer files to sqonk
         user = self.request.user
         if not user.is_authenticated:
-            content = {'Only authenticated users can provide Job overrides'}
+            content: Dict[str, Any] = {
+                'error': 'Only authenticated users can provide Job overrides'
+            }
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Override is expected to be a JSON string,
@@ -1937,13 +1940,15 @@ def get(self, request):
 
         user = self.request.user
         if not user.is_authenticated:
-            content = {'Only authenticated users can access squonk jobs'}
+            content: Dict[str, Any] = {
+                'error': 'Only authenticated users can access squonk jobs'
+            }
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Can't use this method if the Squonk2 agent is not configured
         sq2a_rv = _SQ2A.configured()
         if not sq2a_rv.success:
-            content = {f'The Squonk2 Agent is not configured ({sq2a_rv.msg})'}
+            content = {'error': f'The Squonk2 Agent is not configured ({sq2a_rv.msg})'}
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Iterate through each record, for JobRequests that are not 'finished'
@@ -2029,13 +2034,13 @@ def post(self, request):
         # Only authenticated users can create squonk job requests.
         user = self.request.user
         if not user.is_authenticated:
-            content = {'Only authenticated users can run jobs'}
+            content: Dict[str, Any] = {'error': 'Only authenticated users can run jobs'}
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Can't use this method if the Squonk2 agent is not configured
         sq2a_rv = _SQ2A.configured()
         if not sq2a_rv.success:
-            content = {f'The Squonk2 Agent is not configured ({sq2a_rv.msg})'}
+            content = {'error': f'The Squonk2 Agent is not configured ({sq2a_rv.msg})'}
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         # Collect expected API parameters....
@@ -2064,7 +2069,7 @@ def post(self, request):
         )
         sq2a_rv = _SQ2A.can_run_job(sq2a_run_job_params)
         if not sq2a_rv.success:
-            content = {f'You cannot do this ({sq2a_rv.msg})'}
+            content = {'error': f'You cannot do this ({sq2a_rv.msg})'}
             return Response(content, status=status.HTTP_403_FORBIDDEN)
 
         try: