You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2024/02/12 01:22:53 UTC
(airflow) branch main updated: Add d401 support to kubernetes provider (#37301)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f689c3f606 Add d401 support to kubernetes provider (#37301)
f689c3f606 is described below
commit f689c3f606af5edebd52af830b9128b6b252c063
Author: Rohit Gupta <78...@users.noreply.github.com>
AuthorDate: Mon Feb 12 06:52:46 2024 +0530
Add d401 support to kubernetes provider (#37301)
---
airflow/providers/cncf/kubernetes/callbacks.py | 14 +++++++-------
.../cncf/kubernetes/operators/custom_object_launcher.py | 8 ++++----
.../cncf/kubernetes/operators/spark_kubernetes.py | 2 +-
.../cncf/kubernetes/resource_convert/configmap.py | 4 ++--
.../cncf/kubernetes/resource_convert/env_variable.py | 2 +-
.../providers/cncf/kubernetes/resource_convert/secret.py | 4 ++--
pyproject.toml | 6 ------
7 files changed, 17 insertions(+), 23 deletions(-)
diff --git a/airflow/providers/cncf/kubernetes/callbacks.py b/airflow/providers/cncf/kubernetes/callbacks.py
index 4baef440de..e314c14345 100644
--- a/airflow/providers/cncf/kubernetes/callbacks.py
+++ b/airflow/providers/cncf/kubernetes/callbacks.py
@@ -41,7 +41,7 @@ class KubernetesPodOperatorCallback:
@staticmethod
def on_sync_client_creation(*, client: k8s.CoreV1Api, **kwargs) -> None:
- """Callback method called after creating the sync client.
+ """Invoke this callback after creating the sync client.
:param client: the created `kubernetes.client.CoreV1Api` client.
"""
@@ -49,7 +49,7 @@ class KubernetesPodOperatorCallback:
@staticmethod
def on_pod_creation(*, pod: k8s.V1Pod, client: client_type, mode: str, **kwargs) -> None:
- """Callback method called after creating the pod.
+ """Invoke this callback after creating the pod.
:param pod: the created pod.
:param client: the Kubernetes client that can be used in the callback.
@@ -59,7 +59,7 @@ class KubernetesPodOperatorCallback:
@staticmethod
def on_pod_starting(*, pod: k8s.V1Pod, client: client_type, mode: str, **kwargs) -> None:
- """Callback method called when the pod starts.
+ """Invoke this callback when the pod starts.
:param pod: the started pod.
:param client: the Kubernetes client that can be used in the callback.
@@ -69,7 +69,7 @@ class KubernetesPodOperatorCallback:
@staticmethod
def on_pod_completion(*, pod: k8s.V1Pod, client: client_type, mode: str, **kwargs) -> None:
- """Callback method called when the pod completes.
+ """Invoke this callback when the pod completes.
:param pod: the completed pod.
:param client: the Kubernetes client that can be used in the callback.
@@ -79,7 +79,7 @@ class KubernetesPodOperatorCallback:
@staticmethod
def on_pod_cleanup(*, pod: k8s.V1Pod, client: client_type, mode: str, **kwargs):
- """Callback method called after cleaning/deleting the pod.
+ """Invoke this callback after cleaning/deleting the pod.
:param pod: the completed pod.
:param client: the Kubernetes client that can be used in the callback.
@@ -91,7 +91,7 @@ class KubernetesPodOperatorCallback:
def on_operator_resuming(
*, pod: k8s.V1Pod, event: dict, client: client_type, mode: str, **kwargs
) -> None:
- """Callback method called when resuming the `KubernetesPodOperator` from deferred state.
+ """Invoke this callback when resuming the `KubernetesPodOperator` from deferred state.
:param pod: the current state of the pod.
:param event: the returned event from the Trigger.
@@ -102,7 +102,7 @@ class KubernetesPodOperatorCallback:
@staticmethod
def progress_callback(*, line: str, client: client_type, mode: str, **kwargs) -> None:
- """Callback method to process pod container logs.
+ """Invoke this callback to process pod container logs.
:param line: the read line of log.
:param client: the Kubernetes client that can be used in the callback.
diff --git a/airflow/providers/cncf/kubernetes/operators/custom_object_launcher.py b/airflow/providers/cncf/kubernetes/operators/custom_object_launcher.py
index cd0317c08c..6804f5a84c 100644
--- a/airflow/providers/cncf/kubernetes/operators/custom_object_launcher.py
+++ b/airflow/providers/cncf/kubernetes/operators/custom_object_launcher.py
@@ -210,7 +210,7 @@ class CustomObjectLauncher(LoggingMixin):
template_body: str | None = None,
):
"""
- Creates custom object launcher(sparkapplications crd).
+ Create custom object launcher(sparkapplications crd).
:param kube_client: kubernetes client.
"""
@@ -266,7 +266,7 @@ class CustomObjectLauncher(LoggingMixin):
)
def start_spark_job(self, image=None, code_path=None, startup_timeout: int = 600):
"""
- Launches the pod synchronously and waits for completion.
+ Launch the pod synchronously and waits for completion.
:param image: image name
:param code_path: path to the .py file for python and jar file for scala
@@ -315,7 +315,7 @@ class CustomObjectLauncher(LoggingMixin):
return self.pod_spec, self.spark_obj_spec
def spark_job_not_running(self, spark_obj_spec):
- """Tests if spark_obj_spec has not started."""
+ """Test if spark_obj_spec has not started."""
spark_job_info = self.custom_obj_api.get_namespaced_custom_object_status(
group=self.api_group,
version=self.api_version,
@@ -348,7 +348,7 @@ class CustomObjectLauncher(LoggingMixin):
raise AirflowException(f"Spark Job Failed. Status: {waiting_reason}, Error: {waiting_message}")
def delete_spark_job(self, spark_job_name=None):
- """Deletes spark job."""
+ """Delete spark job."""
spark_job_name = spark_job_name or self.spark_obj_spec.get("metadata", {}).get("name")
if not spark_job_name:
self.log.warning("Spark job not found: %s", spark_job_name)
diff --git a/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py b/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py
index 0c177510eb..16e493aac9 100644
--- a/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py
@@ -280,5 +280,5 @@ class SparkKubernetesOperator(KubernetesPodOperator):
self.client.patch_namespaced_pod(pod.metadata.name, pod.metadata.namespace, body)
def dry_run(self) -> None:
- """Prints out the spark job that would be created by this operator."""
+ """Print out the spark job that would be created by this operator."""
print(prune_dict(self.launcher.body, mode="strict"))
diff --git a/airflow/providers/cncf/kubernetes/resource_convert/configmap.py b/airflow/providers/cncf/kubernetes/resource_convert/configmap.py
index ffd92922d6..1f657146bb 100644
--- a/airflow/providers/cncf/kubernetes/resource_convert/configmap.py
+++ b/airflow/providers/cncf/kubernetes/resource_convert/configmap.py
@@ -21,7 +21,7 @@ from kubernetes.client import models as k8s
def convert_configmap(configmap_name) -> k8s.V1EnvFromSource:
"""
- Converts a str into an k8s object.
+ Convert a str into an k8s object.
:param configmap_name: config map name
:return:
@@ -33,7 +33,7 @@ def convert_configmap_to_volume(
configmap_info: dict[str, str],
) -> tuple[list[k8s.V1Volume], list[k8s.V1VolumeMount]]:
"""
- Converts a dictionary of config_map_name and mount_path into k8s volume mount object and k8s volume.
+ Convert a dictionary of config_map_name and mount_path into k8s volume mount object and k8s volume.
:param configmap_info: a dictionary of {config_map_name: mount_path}
:return:
diff --git a/airflow/providers/cncf/kubernetes/resource_convert/env_variable.py b/airflow/providers/cncf/kubernetes/resource_convert/env_variable.py
index f7d48bf32a..db8c5301cb 100644
--- a/airflow/providers/cncf/kubernetes/resource_convert/env_variable.py
+++ b/airflow/providers/cncf/kubernetes/resource_convert/env_variable.py
@@ -23,7 +23,7 @@ from airflow.exceptions import AirflowException
def convert_env_vars(env_vars) -> list[k8s.V1EnvVar]:
"""
- Converts a dictionary of key:value into a list of env_vars.
+ Convert a dictionary of key:value into a list of env_vars.
:param env_vars:
:return:
diff --git a/airflow/providers/cncf/kubernetes/resource_convert/secret.py b/airflow/providers/cncf/kubernetes/resource_convert/secret.py
index 7d0e0f91ad..323725ee1f 100644
--- a/airflow/providers/cncf/kubernetes/resource_convert/secret.py
+++ b/airflow/providers/cncf/kubernetes/resource_convert/secret.py
@@ -21,7 +21,7 @@ from kubernetes.client import models as k8s
def convert_secret(secret_name: str) -> k8s.V1EnvFromSource:
"""
- Converts a str into an k8s object.
+ Convert a str into an k8s object.
:param secret_name:
:return:
@@ -31,7 +31,7 @@ def convert_secret(secret_name: str) -> k8s.V1EnvFromSource:
def convert_image_pull_secrets(image_pull_secrets: str) -> list[k8s.V1LocalObjectReference]:
"""
- Converts a image pull secret name into k8s local object reference.
+ Convert an image pull secret name into k8s local object reference.
:param image_pull_secrets: comma separated string that contains secrets
:return:
diff --git a/pyproject.toml b/pyproject.toml
index ffa54b8d19..d3de16804d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1363,12 +1363,6 @@ combine-as-imports = true
"airflow/providers/airbyte/operators/airbyte.py" = ["D401"]
"airflow/providers/airbyte/sensors/airbyte.py" = ["D401"]
"airflow/providers/airbyte/triggers/airbyte.py" = ["D401"]
-"airflow/providers/cncf/kubernetes/callbacks.py" = ["D401"]
-"airflow/providers/cncf/kubernetes/operators/custom_object_launcher.py" = ["D401"]
-"airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py" = ["D401"]
-"airflow/providers/cncf/kubernetes/resource_convert/configmap.py" = ["D401"]
-"airflow/providers/cncf/kubernetes/resource_convert/env_variable.py" = ["D401"]
-"airflow/providers/cncf/kubernetes/resource_convert/secret.py" = ["D401"]
"airflow/providers/common/io/xcom/backend.py" = ["D401"]
"airflow/providers/databricks/hooks/databricks.py" = ["D401"]
"airflow/providers/databricks/operators/databricks.py" = ["D401"]