feat: ignore @kubernetes for local runs and add priority_class option

[npow]

Summary

• Ignore @kubernetes for local runs: when @kubernetes is defined statically and the user runs locally (python flow.py run with local datastore), the decorator now acts like @resources — providing resource hints without redirecting execution to Kubernetes. Fixes @kubernetes decorator should be ignored when running flows locally #2588.
• Add priority_class parameter: sets priorityClassName on the pod spec for K8s PriorityClass scheduling. Supported in direct K8s jobs, jobsets, and Argo Workflows. Configurable via METAFLOW_KUBERNETES_PRIORITY_CLASS. Fixes Add a priority class option for the kubernetes flow decorator #1752.

Test plan

• test/unit/test_kubernetes_priority_class.py — unit tests for priority_class parameter handling
• Unit test import fix for KubernetesException

🤖 Generated with Claude Code

[greptile-apps]

Greptile Summary

This PR adds two features to the @kubernetes decorator: (1) when @kubernetes is defined statically in source code and the user runs locally (non-cloud datastore), the decorator now silently enters "local mode" and acts like @resources instead of raising an error; (2) a priority_class parameter is added to set priorityClassName on the pod spec, wired through the CLI, direct K8s jobs, jobsets, and Argo Workflows, and configurable via METAFLOW_KUBERNETES_PRIORITY_CLASS. The implementation is consistent across all execution paths and well-covered by new unit tests.

Confidence Score: 5/5

Safe to merge; no P0/P1 issues found

All changed code paths are logically correct. priority_class is consistently threaded through the decorator, CLI, job/jobset builders, and Argo template. The local-mode guard correctly distinguishes statically-defined vs --with-injected decorators and is properly skipped for all K8s-specific lifecycle hooks. The only finding is a missing blank line in the docstring (P2 style).

No files require special attention

Important Files Changed

Filename	Overview
metaflow/plugins/kubernetes/kubernetes_decorator.py	Adds priority_class attribute and _local_mode flag; local-mode early-return guards added to all relevant lifecycle methods; logic is correct
metaflow/plugins/kubernetes/kubernetes_job.py	Replaces TODO comment with priority_class_name=self._kwargs.get("priority_class") in V1PodSpec; straightforward and correct
metaflow/plugins/kubernetes/kubernetes_jobsets.py	Mirrors the same priority_class_name change as kubernetes_job.py for JobSet pods; consistent and correct
metaflow/plugins/argo/argo_workflows.py	Adds priority_class to direct K8s job call and .priority_class_name() to the Argo template spec; new Template.priority_class_name() method guards against None correctly
metaflow/plugins/kubernetes/kubernetes_cli.py	Adds --priority-class Click option and threads it through to launch_job; consistent with existing option pattern
metaflow/plugins/kubernetes/kubernetes.py	Adds priority_class parameter to both create_job and create_jobset and forwards it to the respective spec builders
metaflow/metaflow_config.py	Adds KUBERNETES_PRIORITY_CLASS config variable with None default; follows existing config pattern
test/unit/test_kubernetes.py	Fixes import aliasing (KubernetesException vs KubeUtilsException) and adds thorough local-mode and priority-class unit tests
test/unit/test_kubernetes_priority_class.py	New unit tests for priority_class including pod-spec propagation via mocked Kubernetes client; covers both set and None cases

_{Reviews (1): Last reviewed commit: "fix: unit test KubernetesException impor..." | Re-trigger Greptile}

[greptile-apps]

Missing blank line in docstring

The priority_class parameter block is missing the blank line separator before security_context that all other parameters have. This breaks the visual consistency of the docstring.

Suggested change

	https://kubernetes.io/docs/concepts/scheduling-eviction/pod-priority-preemption/
	security_context: Dict[str, Any], optional, default None
	https://kubernetes.io/docs/concepts/scheduling-eviction/pod-priority-preemption/
	security_context: Dict[str, Any], optional, default None

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!