From 72fdf1e033ad0a5440cd4952e550ce325579d31c Mon Sep 17 00:00:00 2001 From: hirosassa Date: Sun, 10 May 2026 14:28:53 +0900 Subject: [PATCH 1/4] improve type definition of FlattenableItems for better type narrowing --- gokart/gcs_obj_metadata_client.py | 7 +++---- gokart/task.py | 15 +++++++++++---- gokart/utils.py | 4 ++-- 3 files changed, 16 insertions(+), 10 deletions(-) diff --git a/gokart/gcs_obj_metadata_client.py b/gokart/gcs_obj_metadata_client.py index 2e356308..6f846730 100644 --- a/gokart/gcs_obj_metadata_client.py +++ b/gokart/gcs_obj_metadata_client.py @@ -4,9 +4,8 @@ import functools import json import re -from collections.abc import Iterable from logging import getLogger -from typing import Any, Final +from typing import Any, Final, cast from urllib.parse import urlsplit from gokart.gcs_config import GCSConfig @@ -125,10 +124,10 @@ def _get_patched_obj_metadata( @staticmethod def _get_serialized_string(required_task_outputs: FlattenableItems[RequiredTaskOutput]) -> FlattenableItems[str]: if isinstance(required_task_outputs, RequiredTaskOutput): - return required_task_outputs.serialize() + return cast(FlattenableItems[str], required_task_outputs.serialize()) elif isinstance(required_task_outputs, dict): return {k: GCSObjectMetadataClient._get_serialized_string(v) for k, v in required_task_outputs.items()} - elif isinstance(required_task_outputs, Iterable): + elif isinstance(required_task_outputs, list | tuple): return [GCSObjectMetadataClient._get_serialized_string(ro) for ro in required_task_outputs] else: raise TypeError( diff --git a/gokart/task.py b/gokart/task.py index 09350ef4..c6afc517 100644 --- a/gokart/task.py +++ b/gokart/task.py @@ -153,7 +153,9 @@ def output(self) -> FlattenableItems[TargetOnKart]: def requires(self) -> FlattenableItems[TaskOnKart[Any]]: tasks = self.make_task_instance_dictionary() - return tasks or [] # when tasks is empty dict, then this returns empty list. + if tasks: + return cast(FlattenableItems[TaskOnKart[Any]], tasks) + return [] # when tasks is empty dict, then this returns empty list. def make_task_instance_dictionary(self) -> dict[str, TaskOnKart[Any]]: return {key: var for key, var in vars(self).items() if self.is_task_on_kart(var)} @@ -354,9 +356,14 @@ def dump(self, obj: Any, target: None | str | TargetOnKart = None, custom_labels if isinstance(obj, pd.DataFrame) and obj.empty: raise EmptyDumpError() - required_task_outputs = map_flattenable_items( - lambda task: map_flattenable_items(lambda output: RequiredTaskOutput(task_name=task.get_task_family(), output_path=output.path()), task.output()), - self.requires(), + required_task_outputs = cast( + FlattenableItems[RequiredTaskOutput], + map_flattenable_items( + lambda task: map_flattenable_items( + lambda output: RequiredTaskOutput(task_name=task.get_task_family(), output_path=output.path()), task.output() + ), + self.requires(), + ), ) self._get_output_target(target).dump( diff --git a/gokart/utils.py b/gokart/utils.py index f0b19768..e5b8b7a6 100644 --- a/gokart/utils.py +++ b/gokart/utils.py @@ -27,7 +27,7 @@ def add_config(file_path: str) -> None: T = TypeVar('T') -FlattenableItems: TypeAlias = T | Iterable['FlattenableItems[T]'] | dict[str, 'FlattenableItems[T]'] +FlattenableItems: TypeAlias = T | list['FlattenableItems[T]'] | tuple['FlattenableItems[T]', ...] | dict[str, 'FlattenableItems[T]'] def flatten(targets: FlattenableItems[T]) -> list[T]: @@ -76,7 +76,7 @@ def map_flattenable_items(func: Callable[[T], K], items: FlattenableItems[T]) -> return tuple(map_flattenable_items(func, i) for i in items) if isinstance(items, str): return func(items) # type: ignore - if isinstance(items, Iterable): + if isinstance(items, list): return list(map(lambda item: map_flattenable_items(func, item), items)) return func(items) From bea1ae44815bdbaa69c946a15f745a54306c301f Mon Sep 17 00:00:00 2001 From: hirosassa Date: Sun, 10 May 2026 15:08:14 +0900 Subject: [PATCH 2/4] fix: type issue in worker.py --- gokart/worker.py | 28 +++++++++++++++------------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/gokart/worker.py b/gokart/worker.py index 2c7b1286..97b52ff6 100644 --- a/gokart/worker.py +++ b/gokart/worker.py @@ -57,11 +57,11 @@ import luigi.worker from luigi import notifications from luigi.event import Event -from luigi.scheduler import DISABLED, DONE, FAILED, PENDING, UNKNOWN, WORKER_STATE_ACTIVE, WORKER_STATE_DISABLED, RetryPolicy, Scheduler +from luigi.scheduler import WORKER_STATE_ACTIVE, WORKER_STATE_DISABLED, RetryPolicy, Scheduler from luigi.target import Target from luigi.task import DynamicRequirements, Task, flatten from luigi.task_register import TaskClassException, load_task -from luigi.task_status import RUNNING +from luigi.task_status import DISABLED, DONE, FAILED, PENDING, RUNNING, UNKNOWN from gokart.parameter import ExplicitBoolParameter @@ -178,7 +178,7 @@ def _run_get_new_deps(self) -> list[tuple[str, str, dict[str, str]]] | None: if not requires.complete(self.check_complete): # not all requirements are complete, return them which adds them to the tree - new_deps = [(t.task_module, t.task_family, t.to_str_params()) for t in requires.flat_requirements] + new_deps = [(t.task_module, t.task_family, t.to_str_params()) for t in requires.flat_requirements if isinstance(t, Task)] return new_deps # get the next generator result @@ -205,8 +205,9 @@ def run(self) -> None: if self.check_unfulfilled_deps and not _is_external(self.task): missing = [] for dep in self.task.deps(): + dep = cast(Task, dep) if not self.check_complete(dep): - nonexistent_outputs = [output for output in flatten(dep.output()) if not output.exists()] + nonexistent_outputs = [output for output in cast(list[Target], flatten(dep.output())) if not output.exists()] if nonexistent_outputs: missing.append(f'{dep.task_id} ({", ".join(map(str, nonexistent_outputs))})') else: @@ -335,15 +336,15 @@ class gokart_worker(luigi.Config): id: luigi.StrParameter = luigi.StrParameter(default='', description='Override the auto-generated worker_id') ping_interval: luigi.FloatParameter = luigi.FloatParameter( default=1.0, - config_path=dict(section='core', name='worker-ping-interval'), + config_path={'section': 'core', 'name': 'worker-ping-interval'}, ) keep_alive: luigi.BoolParameter = luigi.BoolParameter( default=False, - config_path=dict(section='core', name='worker-keep-alive'), + config_path={'section': 'core', 'name': 'worker-keep-alive'}, ) count_uniques: luigi.BoolParameter = luigi.BoolParameter( default=False, - config_path=dict(section='core', name='worker-count-uniques'), + config_path={'section': 'core', 'name': 'worker-count-uniques'}, description='worker-count-uniques means that we will keep a worker alive only if it has a unique pending task, as well as having keep-alive true', ) count_last_scheduled: luigi.BoolParameter = luigi.BoolParameter( @@ -351,7 +352,7 @@ class gokart_worker(luigi.Config): ) wait_interval: luigi.FloatParameter = luigi.FloatParameter( default=1.0, - config_path=dict(section='core', name='worker-wait-interval'), + config_path={'section': 'core', 'name': 'worker-wait-interval'}, ) wait_jitter: luigi.FloatParameter = luigi.FloatParameter(default=5.0) @@ -359,19 +360,19 @@ class gokart_worker(luigi.Config): max_reschedules: luigi.IntParameter = luigi.IntParameter( default=1, - config_path=dict(section='core', name='worker-max-reschedules'), + config_path={'section': 'core', 'name': 'worker-max-reschedules'}, ) timeout: luigi.IntParameter = luigi.IntParameter( default=0, - config_path=dict(section='core', name='worker-timeout'), + config_path={'section': 'core', 'name': 'worker-timeout'}, ) task_limit: luigi.OptionalIntParameter = luigi.OptionalIntParameter( default=None, - config_path=dict(section='core', name='worker-task-limit'), + config_path={'section': 'core', 'name': 'worker-task-limit'}, ) retry_external_tasks: luigi.BoolParameter = luigi.BoolParameter( default=False, - config_path=dict(section='core', name='retry-external-tasks'), + config_path={'section': 'core', 'name': 'retry-external-tasks'}, description='If true, incomplete external tasks will be retested for completion while Luigi is running.', ) send_failure_email: luigi.BoolParameter = luigi.BoolParameter(default=True, description='If true, send e-mails directly from the workeron failure') @@ -775,11 +776,12 @@ def _add(self, task: Task, is_complete: bool) -> Generator[Task, None, None]: if deps: for d in deps: + d = cast(Task, d) self._validate_dependency(d) task.trigger_event(Event.DEPENDENCY_DISCOVERED, task, d) yield d # return additional tasks to add - deps = [d.task_id for d in deps] + deps = [cast(Task, d).task_id for d in deps] self._scheduled_tasks[task.task_id] = task self._add_task( From fc0a24ec46da855709a899e0e40dfff54dad5794 Mon Sep 17 00:00:00 2001 From: hirosassa Date: Mon, 11 May 2026 07:39:37 +0900 Subject: [PATCH 3/4] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- gokart/worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gokart/worker.py b/gokart/worker.py index 97b52ff6..09488d8b 100644 --- a/gokart/worker.py +++ b/gokart/worker.py @@ -375,7 +375,7 @@ class gokart_worker(luigi.Config): config_path={'section': 'core', 'name': 'retry-external-tasks'}, description='If true, incomplete external tasks will be retested for completion while Luigi is running.', ) - send_failure_email: luigi.BoolParameter = luigi.BoolParameter(default=True, description='If true, send e-mails directly from the workeron failure') + send_failure_email: luigi.BoolParameter = luigi.BoolParameter(default=True, description='If true, send e-mails directly from the worker on failure') no_install_shutdown_handler: luigi.BoolParameter = luigi.BoolParameter( default=False, description='If true, the SIGUSR1 shutdown handler willNOT be install on the worker' ) From 0170810840a5d95ea6e4bb6e2dc4654133c95fa5 Mon Sep 17 00:00:00 2001 From: hirosassa Date: Fri, 15 May 2026 10:16:30 +0900 Subject: [PATCH 4/4] fix: cast tasks --- gokart/worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gokart/worker.py b/gokart/worker.py index 09488d8b..2659b8ef 100644 --- a/gokart/worker.py +++ b/gokart/worker.py @@ -178,7 +178,7 @@ def _run_get_new_deps(self) -> list[tuple[str, str, dict[str, str]]] | None: if not requires.complete(self.check_complete): # not all requirements are complete, return them which adds them to the tree - new_deps = [(t.task_module, t.task_family, t.to_str_params()) for t in requires.flat_requirements if isinstance(t, Task)] + new_deps = [(t.task_module, t.task_family, t.to_str_params()) for t in cast(list[Task], requires.flat_requirements)] return new_deps # get the next generator result