YDA-5992: add type annotations

.github/workflows/python.yml

@@ -19,12 +19,16 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          python -m pip install flake8==6.0.0 flake8-import-order==0.18.2 darglint==1.8.1 codespell types-requests
+          python -m pip install flake8==6.0.0 flake8-import-order==0.18.2 darglint==1.8.1 codespell mypy types-requests
 
       - name: Lint with flake8
         run: |
           flake8 --statistics
 
+      - name: Check static typing
+        run: |
+          mypy . --explicit-package-bases
+
       - name: Check code for common misspellings
         run: |
           codespell -q 3 --skip="*.r,*.xsd,*.json" || true

admin.py

@@ -11,7 +11,7 @@
 
 
 @api.make()
-def api_admin_has_access(ctx):
+def api_admin_has_access(ctx: rule.Context) -> api.Result:
     """
     Checks if the user has admin access based on user rights or membership in admin-priv group.
 

browse.py

@@ -5,6 +5,7 @@
 
 import re
 from collections import OrderedDict
+from typing import Dict
 
 import magic
 from genquery import AS_DICT, Query
@@ -18,13 +19,13 @@
 
 
 @api.make()
-def api_browse_folder(ctx,
-                      coll='/',
-                      sort_on='name',
-                      sort_order='asc',
-                      offset=0,
-                      limit=10,
-                      space=pathutil.Space.OTHER.value):
+def api_browse_folder(ctx: rule.Context,
+                      coll: str = '/',
+                      sort_on: str = 'name',
+                      sort_order: str = 'asc',
+                      offset: int = 0,
+                      limit: int = 10,
+                      space: str = pathutil.Space.OTHER.value) -> api.Result:
     """Get paginated collection contents, including size/modify date information.
 
     :param ctx:        Combined type of a callback and rei struct
@@ -37,7 +38,7 @@ def api_browse_folder(ctx,
 
     :returns: Dict with paginated collection contents
     """
-    def transform(row):
+    def transform(row: Dict) -> Dict:
         # Remove ORDER_BY etc. wrappers from column names.
         x = {re.sub(r'.*\((.*)\)', '\\1', k): v for k, v in row.items()}
         if 'DATA_NAME' in x and 'META_DATA_ATTR_VALUE' in x:
@@ -104,13 +105,13 @@ def transform(row):
 
 
 @api.make()
-def api_browse_collections(ctx,
-                           coll='/',
-                           sort_on='name',
-                           sort_order='asc',
-                           offset=0,
-                           limit=10,
-                           space=pathutil.Space.OTHER.value):
+def api_browse_collections(ctx: rule.Context,
+                           coll: str = '/',
+                           sort_on: str = 'name',
+                           sort_order: str = 'asc',
+                           offset: int = 0,
+                           limit: int = 10,
+                           space: str = pathutil.Space.OTHER.value) -> api.Result:
     """Get paginated collection contents, including size/modify date information.
 
     This function browses a folder and only looks at the collections in it. No dataobjects.
@@ -126,7 +127,7 @@ def api_browse_collections(ctx,
 
     :returns: Dict with paginated collection contents
     """
-    def transform(row):
+    def transform(row: Dict) -> Dict:
         # Remove ORDER_BY etc. wrappers from column names.
         x = {re.sub(r'.*\((.*)\)', '\\1', k): v for k, v in row.items()}
 
@@ -184,13 +185,13 @@ def transform(row):
 
 
 @api.make()
-def api_search(ctx,
-               search_string,
-               search_type='filename',
-               sort_on='name',
-               sort_order='asc',
-               offset=0,
-               limit=10):
+def api_search(ctx: rule.Context,
+               search_string: str,
+               search_type: str = 'filename',
+               sort_on: str = 'name',
+               sort_order: str = 'asc',
+               offset: int = 0,
+               limit: int = 10) -> api.Result:
     """Get paginated search results, including size/modify date/location information.
 
     :param ctx:           Combined type of a callback and rei struct
@@ -203,7 +204,7 @@ def api_search(ctx,
 
     :returns: Dict with paginated search results
     """
-    def transform(row):
+    def transform(row: Dict) -> Dict:
         # Remove ORDER_BY etc. wrappers from column names.
         x = {re.sub(r'.*\((.*)\)', '\\1', k): v for k, v in row.items()}
 
@@ -216,15 +217,16 @@ def transform(row):
                     'type':        'data',
                     'size':        int(x['DATA_SIZE']),
                     'modify_time': int(x['DATA_MODIFY_TIME'])}
-
-        if 'COLL_NAME' in x:
+        elif 'COLL_NAME' in x:
             _, _, path, subpath = pathutil.info(x['COLL_NAME'])
             if subpath != '':
                 path = path + "/" + subpath
 
             return {'name':        "/{}".format(path),
                     'type':        'coll',
                     'modify_time': int(x['COLL_MODIFY_TIME'])}
+        else:
+            return {}
 
     # Replace, %, _ and \ since iRODS does not handle those correctly.
     # HdR this can only be done in a situation where search_type is NOT status!
@@ -285,7 +287,7 @@ def transform(row):
                         ('items', datas)])
 
 
-def _filter_vault_deposit_index(row):
+def _filter_vault_deposit_index(row: Dict) -> bool:
     """This internal function filters out index collections in deposit vault collections.
        These collections are used internally by Yoda for indexing data package metadata, and
        should not be displayed.
@@ -302,7 +304,7 @@ def _filter_vault_deposit_index(row):
 
 
 @api.make()
-def api_load_text_obj(ctx, file_path='/'):
+def api_load_text_obj(ctx: rule.Context, file_path: str = '/') -> api.Result:
     """Retrieve a text file (as a string) in either the research, deposit, or vault space.
 
     :param ctx:       Combined type of a callback and rei struct

data_access_token.py

@@ -7,6 +7,7 @@
 import secrets
 from datetime import datetime, timedelta
 from traceback import print_exc
+from typing import List
 
 from pysqlcipher3 import dbapi2 as sqlite3
 
@@ -19,15 +20,15 @@
 
 
 @api.make()
-def api_token_generate(ctx, label=None):
+def api_token_generate(ctx: rule.Context, label: str = "") -> api.Result:
     """Generates a token for user authentication.
 
     :param ctx:   Combined type of a callback and rei struct
     :param label: Optional label of the token
 
     :returns: Generated token or API error
     """
-    def generate_token():
+    def generate_token() -> str:
         length = int(config.token_length)
         token = secrets.token_urlsafe(length)
         return token[:length]
@@ -63,14 +64,13 @@ def generate_token():
 
 
 @api.make()
-def api_token_load(ctx):
+def api_token_load(ctx: rule.Context) -> api.Result:
     """Loads valid tokens of user.
 
     :param ctx: Combined type of a callback and rei struct
 
     :returns: Valid tokens
     """
-
     if not token_database_initialized():
         return api.Error('DatabaseError', 'Internal error: token database unavailable')
 
@@ -83,8 +83,8 @@ def api_token_load(ctx):
             conn.execute("PRAGMA key='%s'" % (config.token_database_password))
             for row in conn.execute('''SELECT label, exp_time FROM tokens WHERE user=:user_id AND exp_time > :now''',
                                     {"user_id": user_id, "now": datetime.now()}):
-                exp_time = datetime.strptime(row[1], '%Y-%m-%d %H:%M:%S.%f')
-                exp_time = exp_time.strftime('%Y-%m-%d %H:%M:%S')
+                date_time = datetime.strptime(row[1], '%Y-%m-%d %H:%M:%S.%f')
+                exp_time = date_time.strftime('%Y-%m-%d %H:%M:%S')
                 result.append({"label": row[0], "exp_time": exp_time})
     except Exception:
         print_exc()
@@ -98,7 +98,7 @@ def api_token_load(ctx):
 
 
 @api.make()
-def api_token_delete(ctx, label):
+def api_token_delete(ctx: rule.Context, label: str) -> api.Result:
     """Deletes a token of the user.
 
     :param ctx:   Combined type of a callback and rei struct
@@ -130,10 +130,10 @@ def api_token_delete(ctx, label):
 
 
 @api.make()
-def api_token_delete_expired(ctx):
+def api_token_delete_expired(ctx: rule.Context) -> api.Result:
     """Deletes expired tokens of current user
 
-    :param ctx:   Combined type of a callback and rei struct
+    :param ctx: Combined type of a callback and rei struct
 
     :returns: Status of token deletion
     """
@@ -160,8 +160,9 @@ def api_token_delete_expired(ctx):
     return result
 
 
-def get_all_tokens(ctx):
+def get_all_tokens(ctx: rule.Context) -> List:
     """Retrieve all valid tokens.
+
     :param ctx: Combined type of a callback and rei struct
 
     :returns: Valid tokens
@@ -192,7 +193,7 @@ def get_all_tokens(ctx):
     return result
 
 
-def token_database_initialized():
+def token_database_initialized() -> bool:
     """Checks whether token database has been initialized
 
     :returns: Boolean value

datacite.py

@@ -5,13 +5,14 @@
 
 import random
 import string
+from typing import Dict
 
 import requests
 
 from util import *
 
 
-def metadata_post(ctx, payload):
+def metadata_post(payload: Dict) -> int:
     """Register DOI metadata with DataCite."""
     url = "{}/dois".format(config.datacite_rest_api_url)
     auth = (config.datacite_username, config.datacite_password)
@@ -27,7 +28,7 @@ def metadata_post(ctx, payload):
     return response.status_code
 
 
-def metadata_put(ctx, doi, payload):
+def metadata_put(doi: str, payload: str) -> int:
     """Update metadata with DataCite."""
     url = "{}/dois/{}".format(config.datacite_rest_api_url, doi)
     auth = (config.datacite_username, config.datacite_password)
@@ -43,7 +44,7 @@ def metadata_put(ctx, doi, payload):
     return response.status_code
 
 
-def metadata_get(ctx, doi):
+def metadata_get(doi: str) -> int:
     """Check with DataCite if DOI is available."""
     url = "{}/dois/{}".format(config.datacite_rest_api_url, doi)
     auth = (config.datacite_username, config.datacite_password)
@@ -58,7 +59,7 @@ def metadata_get(ctx, doi):
     return response.status_code
 
 
-def generate_random_id(ctx, length):
+def generate_random_id(length: int) -> str:
     """Generate random ID for DOI."""
     characters = string.ascii_uppercase + string.digits
     return ''.join(random.choice(characters) for x in range(int(length)))