Skip to content

Fix MyPy type errors in providers in cncf/kubernetes provider #57563

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Nov 4, 2025
+9 −4
Merged

Fix MyPy type errors in providers in cncf/kubernetes provider #57563

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
be63da8
Remove SQLA 1 limit in Fab provider
vincbeck Sep 29, 2025
97e79b4
fix mypy errors in cncf-kubernetes-provider
Prab-27 Oct 30, 2025
ca1ac8e
resolved conflicts
Prab-27 Oct 30, 2025
63add54
use filter_for_tis as variable
Prab-27 Oct 30, 2025
46ce2fe
remove cast and use new variable
Prab-27 Nov 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions ...rs/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -505,7 +505,11 @@ def _change_state(
if state is None:
from airflow.models.taskinstance import TaskInstance

state = session.scalar(select(TaskInstance.state).where(TaskInstance.filter_for_tis([key])))
filter_for_tis = TaskInstance.filter_for_tis([key])
if filter_for_tis is not None:
state = session.scalar(select(TaskInstance.state).where(filter_for_tis))
else:
state = None
state = TaskInstanceState(state) if state else None

self.event_buffer[key] = state, None
Expand All @@ -515,7 +519,8 @@ def _get_pod_namespace(ti: TaskInstance):
pod_override = ti.executor_config.get("pod_override")
namespace = None
with suppress(Exception):
namespace = pod_override.metadata.namespace
if pod_override is not None:
namespace = pod_override.metadata.namespace
return namespace or conf.get("kubernetes_executor", "namespace")

def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
Expand Down Expand Up @@ -569,7 +574,7 @@ def try_adopt_task_instances(self, tis: Sequence[TaskInstance]) -> Sequence[Task
tis_to_flush_by_key = {ti.key: ti for ti in tis if ti.queued_by_job_id}
kube_client: client.CoreV1Api = self.kube_client
for scheduler_job_id in scheduler_job_ids:
scheduler_job_id = self._make_safe_label_value(str(scheduler_job_id))
scheduler_job_id_safe_label = self._make_safe_label_value(str(scheduler_job_id))
# We will look for any pods owned by the no-longer-running scheduler,
# but will exclude only successful pods, as those TIs will have a terminal state
# and not be up for adoption!
Expand All @@ -579,7 +584,7 @@ def try_adopt_task_instances(self, tis: Sequence[TaskInstance]) -> Sequence[Task
"field_selector": "status.phase!=Succeeded",
"label_selector": (
"kubernetes_executor=True,"
f"airflow-worker={scheduler_job_id},{POD_EXECUTOR_DONE_KEY}!=True"
f"airflow-worker={scheduler_job_id_safe_label},{POD_EXECUTOR_DONE_KEY}!=True"
),
}
pod_list = self._list_pods(query_kwargs)
Expand Down