Improve static typing and mypy compatibility

- Update type annotations across multiple scripts
- Add explicit type casts where needed
- Refactor functions to pass mypy strict mode
- Add mypy.ini basic configuration for Python 3.11

Author: pabo99

mypi.ini

@@ -0,0 +1,3 @@
+[mypy]
+python_version = 3.11
+strict = True

stuff/lint.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+set -e
+flake8 utils/
+mypy utils/

utils/Pipfile

@@ -9,7 +9,7 @@ name = "pypi"
 
 mypy = ">=0.782"
 pycodestyle = ">=2.6.0"
-
+flake8 = ">=3.9.2"
 
 [packages]
 

utils/container.py

@@ -5,7 +5,7 @@
 import os.path
 
 from types import TracebackType
-from typing import AnyStr, Iterator, IO, Optional, Type, Sequence
+from typing import Iterator, IO, Optional, Type, Sequence
 
 import problems
 
@@ -16,7 +16,7 @@
 
 @contextlib.contextmanager
 def _maybe_open(path: Optional[str],
-                mode: str) -> Iterator[Optional[IO[AnyStr]]]:
+                mode: str) -> Iterator[Optional[IO[str]]]:
     """A contextmanager that can open a file, or return None.
 
     This is useful to provide arguments to subprocess.call() and its friends.

utils/download_and_sync_courses.py

@@ -5,7 +5,7 @@
 import logging
 import os
 import zipfile
-from typing import Dict, Any
+from typing import Dict, Any, List, NamedTuple, cast
 from urllib.parse import urlparse, urljoin
 import omegaup.api
 import shutil
@@ -18,8 +18,6 @@
 logging.basicConfig(level=logging.INFO)
 LOG = logging.getLogger(__name__)
 
-API_CLIENT = None
-BASE_URL = None
 
 # 👇 Add your course aliases here
 COURSE_ALIASES = [
@@ -31,11 +29,15 @@
     os.path.join(os.path.dirname(__file__), "..", "Courses"))
 
 
-def handle_input():
-    global BASE_URL, API_TOKEN
+class InputArgs(NamedTuple):
+    api_token: str
+    base_url: str
+
+
+def handle_input() -> InputArgs:
     parser = argparse.ArgumentParser(
-        description=f"Download and extract problems from multiple course"
-                    f" assignments"
+        description="Download and extract problems from multiple course"
+                    " assignments"
     )
     parser.add_argument("--url",
                         default="https://omegaup.com",
@@ -45,23 +47,33 @@ def handle_input():
                         default=os.environ.get("OMEGAUP_API_TOKEN"),
                         required=("OMEGAUP_API_TOKEN" not in os.environ))
     args = parser.parse_args()
-    BASE_URL = args.url
-    return args.api_token
+    if args.api_token is None:
+        parser.error(
+            "API token is required (use --api-token or set OMEGAUP_API_TOKEN)")
+    return InputArgs(api_token=str(args.api_token), base_url=str(args.url))
 
 
-def get_json(endpoint: str, params: Dict[str, str]) -> Dict[str, Any]:
-    return API_CLIENT.query(endpoint, params)
+def get_json(
+        client: omegaup.api.Client,
+        endpoint: str,
+        params: Dict[str, str],
+        base_url: str
+) -> Dict[str, Any]:
+    return cast(Dict[str, Any], client.query(endpoint, params))
 
 
 def sanitize_filename(name: str) -> str:
     return "".join(c for c in name if c.isalnum() or c in " -_").strip()
 
 
 def get_course_details(
+        client: omegaup.api.Client,
         course_alias: str,
-        course_base_folder: str
+        course_base_folder: str,
+        base_url: str
 ) -> Dict[str, Any]:
-    details = get_json("/api/course/details/", {"alias": course_alias})
+    params = {"alias": course_alias}
+    details = get_json(client, "/api/course/details/", params, base_url)
     details.pop("assignments", None)
     details.pop("clarifications", None)
 
@@ -76,29 +88,49 @@ def get_course_details(
     return details
 
 
-def get_assignments(course_alias: str):
-    return get_json("/api/course/listAssignments/",
-                    {"course_alias": course_alias})["assignments"]
+def get_assignments(
+        client: omegaup.api.Client,
+        course_alias: str,
+        base_url: str
+) -> List[Dict[str, Any]]:
+    endpoint = "/api/course/listAssignments/"
+    params = {"course_alias": course_alias}
+    return cast(
+        List[Dict[str, Any]],
+        get_json(client, endpoint, params, base_url)["assignments"]
+    )
 
 
-def get_assignment_details(course_alias: str, assignment_alias: str):
-    return get_json("/api/course/assignmentDetails/", {
+def get_assignment_details(
+        client: omegaup.api.Client,
+        course_alias: str,
+        assignment_alias: str,
+        base_url: str
+) -> Dict[str, Any]:
+    endpoint = "/api/course/assignmentDetails/"
+    params = {
         "course": course_alias,
         "assignment": assignment_alias
-    })
+    }
+    return get_json(client, endpoint, params, base_url)
 
 
-def download_and_unzip(problem_alias: str, assignment_folder: str) -> bool:
+def download_and_unzip(client: omegaup.api.Client, problem_alias: str,
+                       assignment_folder: str, base_url: str) -> bool:
     try:
         download_url = urljoin(
-            BASE_URL,
+            base_url,
             f"/api/problem/download/problem_alias/{problem_alias}/"
         )
         parsed_url = urlparse(download_url)
+        if parsed_url.hostname is None:
+            LOG.error(f"Invalid download URL (missing hostname): "
+                      f"{download_url}")
+            return False
         conn = http.client.HTTPSConnection(parsed_url.hostname,
                                            context=context)
 
-        headers = {'Authorization': f'token {API_CLIENT.api_token}'}
+        headers = {'Authorization': f'token {client.api_token}'}
         path = parsed_url.path
 
         conn.request("GET", path, headers=headers)
@@ -171,10 +203,9 @@ def download_and_unzip(problem_alias: str, assignment_folder: str) -> bool:
         return False
 
 
-def main():
-    global API_CLIENT
-    api_token = handle_input()
-    API_CLIENT = omegaup.api.Client(api_token=api_token, url=BASE_URL)
+def main() -> None:
+    input = handle_input()
+    client = omegaup.api.Client(api_token=input.api_token, url=input.base_url)
 
     if os.path.exists(BASE_COURSE_FOLDER):
         LOG.warning("Delete existing course folder to avoid conflicts")
@@ -186,7 +217,7 @@ def main():
     for course_alias in COURSE_ALIASES:
         LOG.info(f"📘 Starting course: {course_alias}")
         try:
-            assignments = get_assignments(course_alias)
+            assignments = get_assignments(client, course_alias, input.base_url)
 
             if not assignments:
                 LOG.warning(f"No assignments found in {course_alias}.")
@@ -203,8 +234,12 @@ def main():
                 )
 
                 try:
-                    details = get_assignment_details(course_alias,
-                                                     assignment_alias)
+                    details = get_assignment_details(
+                        client,
+                        course_alias,
+                        assignment_alias,
+                        input.base_url
+                    )
                     assignment_folder = os.path.join(course_folder,
                                                      assignment_alias)
                     os.makedirs(assignment_folder, exist_ok=True)
@@ -213,8 +248,10 @@ def main():
 
                     for problem in problems:
                         try:
-                            downloaded = download_and_unzip(problem["alias"],
-                                                            assignment_folder)
+                            downloaded = download_and_unzip(client,
+                                                            problem["alias"],
+                                                            assignment_folder,
+                                                            input.base_url)
                             if downloaded:
                                 rel_path = os.path.join(
                                     "Courses",

utils/generateresources.py

@@ -2,7 +2,6 @@
 import argparse
 import concurrent.futures
 import datetime
-import json
 import logging
 import os
 import re

utils/runtests.py

@@ -2,7 +2,6 @@
 import argparse
 import collections
 import concurrent.futures
-import decimal
 import json
 import logging
 import os
@@ -289,9 +288,6 @@ def _main() -> None:
             failureMessages: DefaultDict[
                 str, List[str]] = collections.defaultdict(list)
 
-            normalizedScore = decimal.Decimal(got.get('score', 0))
-            scaledScore = round(normalizedScore, 15) * 100
-
             if testResult['state'] != 'passed':
                 # Build a table that reports groups and case verdicts.
                 groupReportTable = [

utils/update_assignment_problems.py

@@ -5,8 +5,9 @@
 import logging
 import os
 import datetime
-from typing import Dict, Any, List, NamedTuple, Tuple
+from typing import Dict, Any, List, NamedTuple, Sequence, Tuple
 import omegaup.api
+from omegaup.api import _CourseAssignment
 import re
 from urllib.parse import urlparse, urljoin
 import shutil
@@ -67,7 +68,10 @@ def handle_input() -> Tuple[str, str, str]:
     return args.api_token, args.url, args.input
 
 
-def assignment_exists(assignments: List[Any], alias: str) -> bool:
+def assignment_exists(
+        assignments: Sequence[_CourseAssignment],
+        alias: str
+) -> bool:
     return any(a.alias == alias for a in assignments)
 
 
@@ -91,8 +95,8 @@ def create_assignment(
             assignment_type="homework",
             name=assignment_alias,
             description=f"Auto-created assignment {assignment_alias}",
-            start_time=int(now.timestamp()),
-            finish_time=int(finish.timestamp()),
+            start_time=now,
+            finish_time=finish,
             unlimited_duration=True
         )
         LOG.info(f"✅ Created assignment '{assignment_alias}'")
@@ -112,6 +116,10 @@ def download_and_unzip(
             f"/api/problem/download/problem_alias/{problem_alias}/"
         )
         parsed_url = urlparse(download_url)
+        if parsed_url.hostname is None:
+            LOG.error(f"Invalid download URL (missing hostname): "
+                      f"{download_url}")
+            return False
         conn = http.client.HTTPSConnection(parsed_url.hostname,
                                            context=context)
 
@@ -191,10 +199,10 @@ def download_and_unzip(
 
 def process_add(
         data: Dict[str, Any],
-        problems_data: Dict[str, List[Dict[str, str]]],
+        problems_data: Dict[str, List[ProblemEntry]],
         client: omegaup.api.Client,
         base_url: str
-):
+) -> None:
     for item in data.get("add_problem", []):
         course = item["course_alias"]
         assignment = item["assignment_alias"]
@@ -238,6 +246,11 @@ def process_add(
             os.makedirs(assignment_folder, exist_ok=True)
 
             LOG.info(f"📥 Downloading and unzipping problem '{problem}'")
+
+            if client.api_token is None:
+                LOG.error("❌ API token is required.")
+                return
+
             success = download_and_unzip(
                 problem_alias=problem,
                 assignment_folder=assignment_folder,
@@ -262,9 +275,9 @@ def process_add(
 
 def process_remove(
        data: Dict[str, Any],
-       problems_data: Dict[str, List[Dict[str, str]]],
+       problems_data: Dict[str, List[ProblemEntry]],
        client: omegaup.api.Client
-):
+) -> None:
     for item in data.get("remove_problem", []):
         course = item["course_alias"]
         assignment = item["assignment_alias"]
@@ -342,7 +355,9 @@ def load_problems_json() -> Dict[str, List[ProblemEntry]]:
     return {"problems": []}
 
 
-def save_problems_json(data: Dict[str, List[ProblemEntry]]):
+def save_problems_json(data: Dict[str, List[ProblemEntry]]) -> None:
+    """Saves the problems data to the problems.json file."""
+
     with open(PROBLEMS_JSON_PATH, "w", encoding="utf-8") as f:
         json.dump(
             {"problems": [p._asdict() for p in data["problems"]]},
@@ -352,15 +367,19 @@ def save_problems_json(data: Dict[str, List[ProblemEntry]]):
 
 
 def add_problem_to_json(course: str, assignment: str, problem: str,
-                        problems_data: Dict[str, List[ProblemEntry]]):
+                        problems_data: Dict[str, List[ProblemEntry]]) -> None:
     path = f"Courses/{course}/{assignment}/{problem}"
     if not any(p.path == path for p in problems_data["problems"]):
         problems_data["problems"].append(ProblemEntry(path=path))
         LOG.info(f"📝 Added '{path}' to problems.json")
 
 
-def remove_problem_from_json(course: str, assignment: str, problem: str,
-                             problems_data: Dict[str, List[ProblemEntry]]):
+def remove_problem_from_json(
+        course: str,
+        assignment: str,
+        problem: str,
+        problems_data: Dict[str, List[ProblemEntry]]
+) -> None:
     path = f"Courses/{course}/{assignment}/{problem}"
     before = len(problems_data["problems"])
     problems_data["problems"] = [
@@ -370,7 +389,7 @@ def remove_problem_from_json(course: str, assignment: str, problem: str,
         LOG.info(f"🗑️  Removed '{path}' from problems.json")
 
 
-def main():
+def main() -> None:
     api_token, base_url, input_path = handle_input()
     client = omegaup.api.Client(api_token=api_token, url=base_url)
 
@@ -401,7 +420,7 @@ def main():
     except (IOError, json.JSONDecodeError) as e:
         LOG.info(
             f"🧹 Cleared 'add_problem' and 'remove_problem' arrays in "
-            f"{input_path}"
+            f"{input_path}: {e}"
         )