Add missing type annotations

app.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-__copyright__ = 'Copyright (c) 2021-2023, Utrecht University'
+__copyright__ = 'Copyright (c) 2021-2024, Utrecht University'
 __license__   = 'GPLv3, see LICENSE'
 
 import json
@@ -178,15 +178,16 @@ def static_loader() -> Optional[Response]:
 
     :returns: Static file
     """
-    result = get_validated_static_path(
+    static_dir, asset_name = get_validated_static_path(
         request.full_path,
         request.path,
         app.config.get('YODA_THEME_PATH'),
         app.config.get('YODA_THEME')
     )
-    if result is not None:
-        static_dir, asset_name = result
+    if static_dir and asset_name:
         return send_from_directory(static_dir, asset_name)
+    else:
+        return None
 
 
 @app.before_request

connman.py

@@ -14,20 +14,20 @@
 
 
 class Session(object):
-    def __init__(self, sid: int, irods: iRODSSession) -> None:
+    def __init__(self, sid: str, irods: iRODSSession) -> None:
         """Session object storing the iRODS session object.
 
         :param sid:   Flask session identifier
         :param irods: iRODS session
         """
-        self.sid: int             = sid               # Flask session identifier
+        self.sid: str             = sid               # Flask session identifier
         self.time: float          = time.time()       # Flask session start time
         self.irods: iRODSSession  = irods             # iRODS session
         self.irods_time: float    = time.time()       # iRODS session start time
         self.lock: threading.Lock = threading.Lock()
 
 
-sessions: Dict[int, Session] = dict()  # Custom session dict instead of Flask session (cannot pickle iRODS session)
+sessions: Dict[str, Session] = dict()  # Custom session dict instead of Flask session (cannot pickle iRODS session)
 lock: threading.Lock = threading.Lock()
 
 
@@ -54,7 +54,7 @@ def gc() -> None:
 gc_thread.start()
 
 
-def get(sid: int) -> Optional[iRODSSession]:
+def get(sid: str) -> Optional[iRODSSession]:
     """Retrieve iRODS session object from session."""
     if sid in sessions:
         s: Session = sessions[sid]
@@ -65,7 +65,7 @@ def get(sid: int) -> Optional[iRODSSession]:
     return s.irods
 
 
-def add(sid: int, irods: iRODSSession) -> None:
+def add(sid: str, irods: iRODSSession) -> None:
     """Add Flask sid and iRODS session to our custom session dict.
 
     :param sid:   Flask session identifier
@@ -80,7 +80,7 @@ def add(sid: int, irods: iRODSSession) -> None:
     print(f"[login]: Successfully connected to iRODS for session {sid}'")
 
 
-def release(sid: int) -> None:
+def release(sid: str) -> None:
     """Release a session.
 
     :param sid: Flask session identifier
@@ -93,7 +93,7 @@ def release(sid: int) -> None:
         s.lock.release()
 
 
-def clean(sid: int) -> None:
+def clean(sid: str) -> None:
     """Clean a session.
 
     :param sid: Flask session identifier
@@ -106,7 +106,7 @@ def clean(sid: int) -> None:
         print(f"[logout]: Cleanup session {sid}")
 
 
-def extend(sid: int) -> None:
+def extend(sid: str) -> None:
     """Extend session TTLs.
 
     :param sid: Flask session identifier

research/research.py

@@ -8,14 +8,15 @@
 import queue
 import threading
 import urllib.parse
-from typing import Iterator
+from typing import Iterator, Optional
 
 from flask import (
     abort, Blueprint, current_app as app, g, jsonify, make_response,
     render_template, request, Response, session, stream_with_context
 )
 from irods.data_object import iRODSDataObject
 from irods.exception import CAT_NO_ACCESS_PERMISSION, CAT_NO_ROWS_FOUND
+from irods.manager.data_object_manager import DataObjectManager
 from irods.message import iRODSMessage
 
 import api
@@ -29,24 +30,24 @@
 
 
 class Chunk:
-    def __init__(self, data_objects, path: str, number: int, size: int, data, resource: str) -> None:
-        self.data_objects = data_objects
-        self.path = path
-        self.number = number
-        self.size = size
-        self.data = data
-        self.resource = resource
+    def __init__(self: Chunk, data_objects: Optional[DataObjectManager], path: Optional[str], number: int, size: int, data: Optional[str], resource: Optional[str]) -> None:
+        self.data_objects: Optional[DataObjectManager] = data_objects
+        self.path: Optional[str] = path
+        self.number: int = number
+        self.size: int = size
+        self.data: Optional[str] = data
+        self.resource: Optional[str] = resource
 
 
-q: queue.Queue = queue.Queue(4)
-r: queue.Queue = queue.Queue(1)
+q: queue.Queue[Chunk] = queue.Queue(4)
+r: queue.Queue[bool] = queue.Queue(1)
 
 
 def irods_writer() -> None:
     failure = False
     while True:
         chunk = q.get()
-        if chunk.path:
+        if chunk.path and chunk.data_objects is not None:
             if not failure:
                 try:
                     with chunk.data_objects.open(chunk.path, 'a', chunk.resource) as obj_desc:

setup.cfg

@@ -12,7 +12,7 @@ ignore_missing_imports = True
 warn_unreachable = True
 no_implicit_optional = True
 check_untyped_defs = False
-disallow_any_generics = True
+disallow_any_generics = False
 disallow_incomplete_defs = True
 disallow_untyped_calls = False
 disallow_untyped_defs = False

util.py

@@ -90,7 +90,7 @@ def unicode_secure_filename(filename: str) -> str:
 
 def get_validated_static_path(
     full_path: str, request_path: str, yoda_theme_path: str, yoda_theme: str
-) -> Optional[Tuple[str, str]]:
+) -> Tuple[str, str]:
     """
     Static files handling - recognisable through '/assets/'
     Confirms that input path is valid and return corresponding static path
@@ -112,13 +112,13 @@ def get_validated_static_path(
         _, asset_name = path.split(request_path)
         # Make sure asset_name is safe
         if asset_name != secure_filename(asset_name):
-            return None
+            return "", ""
 
         if parts[0] == "assets":
             # Main assets
             static_dir = safe_join(user_static_area + "/static", *parts[1:])
             if not static_dir:
-                return None
+                return "", ""
             user_static_filename = path.join(static_dir, asset_name)
             if not path.exists(user_static_filename):
                 static_dir = safe_join("/var/www/yoda/static", *parts[1:])
@@ -127,47 +127,52 @@ def get_validated_static_path(
             module = parts[0]
             # Make sure module name is safe
             if module != secure_filename(module):
-                return None
+                return "", ""
 
             module_static_area = path.join(module, "static", module)
             user_static_filename = safe_join(
                 path.join(user_static_area, module_static_area), *parts[2:], asset_name
             )
             if not user_static_filename:
-                return None
+                return "", ""
 
             if path.exists(user_static_filename):
                 static_dir = path.join(user_static_area, module_static_area, *parts[2:])
             else:
                 static_dir = path.join("/var/www/yoda/", module_static_area, *parts[2:])
 
         full_path = path.join(static_dir, asset_name)
+
         # Check that path is correct
         if path.exists(full_path):
             return static_dir, asset_name
 
+    return "", ""
+
 
 def length_check(message: str) -> Tuple[str, bool]:
     """
     Check banner message length.
 
-    :param message: Message to validate.
-    :returns: Error message and validity status.
+    :param message: Message to validate
+
+    :returns: Error message and validity status
     """
     max_length = 256
     if not message:
         return "Empty banner message found.", False
     elif len(message) > max_length:
         return "Banner message too long.", False
-    return None, True
+    return "", True
 
 
 def get_theme_directories(theme_path: str) -> List[str]:
     """
     Function to retrieve theme directory names in the specified path, sorted alphabetically.
 
-    :param theme_path: The path where theme directories are located.
-    :returns: A sorted list of directory names including 'uu', or an empty list in case of an error.
+    :param theme_path: The path where theme directories are located
+
+    :returns: A sorted list of directory names including 'uu', or an empty list in case of an error
     """
     try:
         directories = [name for name in listdir(theme_path) if path.isdir(path.join(theme_path, name))] + ['uu']