fix up typing

Author: nsaje

pyproject.toml

@@ -1,20 +1,3 @@
-[tool.black]
-line-length = 79
-exclude = '''
-/(
-    \.git
-)/
-'''
-
-[tool.isort]
-skip = ['.git', 'venv']
-known_tests = 'tests'
-sections = ['FUTURE', 'STDLIB', 'THIRDPARTY', 'FIRSTPARTY', 'TESTS', 'LOCALFOLDER']
-default_section = 'THIRDPARTY'
-use_parentheses = true
-multi_line_output = 3
-include_trailing_comma = true
-force_grid_wrap = 0
-combine_as_imports = true
-line_length = 79
-float_to_top = true
+[tool.mypy]
+ignore_missing_imports = true
+follow_imports = "skip"

requirements-lint.txt

@@ -1,2 +1,3 @@
+mypy==1.19.0
 ruff==0.14.7
 types-redis

setup.py

@@ -4,7 +4,7 @@
 
 VERSION_FILE = "tasktiger/__init__.py"
 with open(VERSION_FILE, encoding="utf8") as fd:
-    version = re.search(r'__version__ = ([\'"])(.*?)\1', fd.read()).group(2)
+    version = re.search(r'__version__ = ([\'"])(.*?)\1', fd.read()).group(2)  # type: ignore
 
 with open("README.rst", encoding="utf-8") as file:
     long_description = file.read()

tasktiger/runner.py

@@ -42,7 +42,7 @@ def run_eager_task(self, task: "Task") -> None:
         """
         raise NotImplementedError("Eager tasks are not supported.")
 
-    def on_permanent_error(self, task: "Task", execution: Dict[str, Any]) -> None:
+    def on_permanent_error(self, task: "Task", execution: Dict[str, Any] | None) -> None:
         """
         Called if the task fails permanently.
 

tasktiger/task.py

@@ -532,8 +532,9 @@ def tasks_from_queue(
                 for idx, serialized_data, serialized_executions, ts in zip(
                     range(len(items)), results[0], results[1:], tss
                 ):
-                    if serialized_data is None and include_not_found:
-                        data = {"id": items[idx][0]}
+                    if serialized_data is None:
+                        if include_not_found:
+                            data = {"id": items[idx][0]}
                     else:
                         data = json.loads(serialized_data)
 
@@ -554,8 +555,9 @@ def tasks_from_queue(
                     [tiger._key("task", item[0]) for item in items]
                 )
                 for idx, serialized_data, ts in zip(range(len(items)), result, tss):
-                    if serialized_data is None and include_not_found:
-                        data = {"id": items[idx][0]}
+                    if serialized_data is None:
+                        if include_not_found:
+                            data = {"id": items[idx][0]}
                     else:
                         data = json.loads(serialized_data)
 

tasktiger/tasktiger.py

@@ -13,6 +13,7 @@
     Iterable,
     List,
     Optional,
+    ParamSpec,
     Tuple,
     Type,
     TypeVar,
@@ -24,7 +25,6 @@
 import redis
 import structlog
 from structlog.stdlib import BoundLogger
-from typing_extensions import ParamSpec
 
 from ._internal import (
     ACTIVE,