Codestyle enhancements

app/corpora/cqi_over_sio/__init__.py

@@ -6,7 +6,7 @@ from flask_login import current_user
 from flask_socketio import Namespace
 from inspect import signature
 from threading import Lock
-from typing import Callable, Dict, List
+from typing import Callable, Dict, List, Optional
 from app import db, hashids, socketio
 from app.decorators import socketio_login_required
 from app.models import Corpus, CorpusStatus
@@ -92,8 +92,8 @@ class CQiNamespace(Namespace):
 
     @socketio_login_required
     def on_init(self, db_corpus_hashid: str):
-        db_corpus_id = hashids.decode(db_corpus_hashid)
-        db_corpus = Corpus.query.get(db_corpus_id)
+        db_corpus_id: int = hashids.decode(db_corpus_hashid)
+        db_corpus: Optional[Corpus] = Corpus.query.get(db_corpus_id)
         if db_corpus is None:
             return {'code': 404, 'msg': 'Not Found'}
         if not (db_corpus.user == current_user
@@ -112,7 +112,7 @@ class CQiNamespace(Namespace):
             db.session.commit()
         db_corpus.num_analysis_sessions = Corpus.num_analysis_sessions + 1
         db.session.commit()
-        retry_counter = 20
+        retry_counter: int = 20
         while db_corpus.status != CorpusStatus.RUNNING_ANALYSIS_SESSION:
             if retry_counter == 0:
                 db_corpus.num_analysis_sessions = Corpus.num_analysis_sessions - 1
@@ -121,11 +121,15 @@ class CQiNamespace(Namespace):
             socketio.sleep(3)
             retry_counter -= 1
             db.session.refresh(db_corpus)
-        cqi_client = CQiClient(f'cqpserver_{db_corpus_id}', timeout=float('inf'))
-        session['cqi_over_sio'] = {}
-        session['cqi_over_sio']['cqi_client'] = cqi_client
-        session['cqi_over_sio']['cqi_client_lock'] = Lock()
-        session['cqi_over_sio']['db_corpus_id'] = db_corpus_id
+        cqi_client: CQiClient = CQiClient(
+            f'cqpserver_{db_corpus_id}',
+            timeout=float('inf')
+        )
+        session['cqi_over_sio'] = {
+            'cqi_client': cqi_client,
+            'cqi_client_lock': Lock(),
+            'db_corpus_id': db_corpus_id
+        }
         return {'code': 200, 'msg': 'OK'}
 
     @socketio_login_required
@@ -193,7 +197,8 @@ class CQiNamespace(Namespace):
         except (BrokenPipeError, CQiException):
             pass
         cqi_client_lock.release()
-        db_corpus = Corpus.query.get(db_corpus_id)
-        if db_corpus is not None:
-            db_corpus.num_analysis_sessions = Corpus.num_analysis_sessions - 1
-            db.session.commit()
+        db_corpus: Optional[Corpus] = Corpus.query.get(db_corpus_id)
+        if db_corpus is None:
+            return
+        db_corpus.num_analysis_sessions = Corpus.num_analysis_sessions - 1
+        db.session.commit()

app/corpora/cqi_over_sio/utils.py

-from cqi.models.corpora import Corpus
-from cqi.models.subcorpora import Subcorpus
+from cqi.models.corpora import Corpus as CQiCorpus
+from cqi.models.subcorpora import Subcorpus as CQiSubcorpus
 from typing import Dict, List
-from app.models import Corpus
 
 
-def lookups_by_cpos(corpus: Corpus, cpos_list: List[int]) -> Dict:
+def lookups_by_cpos(corpus: CQiCorpus, cpos_list: List[int]) -> Dict:
     lookups = {}
     lookups['cpos_lookup'] = {cpos: {} for cpos in cpos_list}
     for attr in corpus.positional_attributes.list():
-        cpos_attr_values = attr.values_by_cpos(cpos_list)
+        cpos_attr_values: List[str] = attr.values_by_cpos(cpos_list)
         for i, cpos in enumerate(cpos_list):
             lookups['cpos_lookup'][cpos][attr.attrs['name']] = \
                 cpos_attr_values[i]
     for attr in corpus.structural_attributes.list():
         # We only want to iterate over non subattributes, identifiable by
-        # attr.attrs['has_values'] == False
-        if attr.attrs['has_values']:
+        # attr.has_values == False
+        if attr.has_values:
             continue
-        cpos_attr_ids = attr.ids_by_cpos(cpos_list)
+        cpos_attr_ids: List[int] = attr.ids_by_cpos(cpos_list)
         for i, cpos in enumerate(cpos_list):
             if cpos_attr_ids[i] == -1:
                 continue
@@ -28,7 +27,7 @@ def lookups_by_cpos(corpus: Corpus, cpos_list: List[int]) -> Dict:
         subattrs = corpus.structural_attributes.list(filters={'part_of': attr})
         if not subattrs:
             continue
-        lookup_name = f'{attr.attrs["name"]}_lookup'
+        lookup_name: str = f'{attr.attrs["name"]}_lookup'
         lookups[lookup_name] = {}
         for attr_id in occured_attr_ids:
             lookups[lookup_name][attr_id] = {}
@@ -40,7 +39,7 @@ def lookups_by_cpos(corpus: Corpus, cpos_list: List[int]) -> Dict:
 
 
 def partial_export_subcorpus(
-    subcorpus: Subcorpus,
+    subcorpus: CQiSubcorpus,
     match_id_list: List[int],
     context: int = 25
 ) -> Dict:
@@ -89,7 +88,7 @@ def partial_export_subcorpus(
 
 
 def export_subcorpus(
-    subcorpus: Subcorpus,
+    subcorpus: CQiSubcorpus,
     context: int = 25,
     cutoff: float = float('inf'),
     offset: int = 0