refactor: use render_templates from ti and overwrite_rtif_after_execution

Author: insomnes

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/decorators/kubernetes_cmd.py

@@ -22,7 +22,6 @@
 
 from airflow.decorators.base import DecoratedOperator, TaskDecorator, task_decorator_factory
 from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
-from airflow.providers.cncf.kubernetes.template_rendering import refresh_rendered_fields
 from airflow.utils.context import context_merge
 from airflow.utils.operator_helpers import determine_kwargs
 
@@ -38,6 +37,7 @@ class _KubernetesCmdDecoratedOperator(DecoratedOperator, KubernetesPodOperator):
     custom_operator_name = "@task.kubernetes_cmd"
 
     template_fields: Sequence[str] = KubernetesPodOperator.template_fields
+    overwrite_rtif_after_execution: bool = True
 
     def __init__(self, *, python_callable: Callable, args_only: bool = False, **kwargs) -> None:
         self.args_only = args_only
@@ -76,7 +76,7 @@ def execute(self, context: Context):
         else:
             self.cmds = generated
             self.arguments = []
-        refresh_rendered_fields(context["task_instance"])
+        context["ti"].render_templates()  # type: ignore[attr-defined]
         return super().execute(context)
 
     def _generate_cmds(self, context: Context) -> list[str]:

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/template_rendering.py

@@ -77,23 +77,3 @@ def get_rendered_k8s_spec(task_instance: TaskInstance, session: SASession = NEW_
         except (TemplateAssertionError, UndefinedError) as e:
             raise AirflowException(f"Unable to render a k8s spec for this taskinstance: {e}") from e
     return rendered_k8s_spec
-
-
-@provide_session
-def refresh_rendered_fields(task_instance: TaskInstance, session=NEW_SESSION) -> None:
-    """
-    Rewrite the underlying rendered values for a task instance in the metadatabase.
-
-    TaskInstance.get_rendered_template_fields() cannot be used because this will retrieve the
-    RenderedTaskInstanceFields from the metadatabase which doesn't have the runtime-evaluated
-    command value in case of `@task.kubernetes_cmd` decorator.
-    """
-    from airflow.models.renderedtifields import RenderedTaskInstanceFields
-
-    """Update rendered task instance fields for cases where runtime evaluated, not templated."""
-
-    rtif = RenderedTaskInstanceFields(task_instance)
-    RenderedTaskInstanceFields.write(rtif, session=session)
-    RenderedTaskInstanceFields.delete_old_records(
-        task_instance.task_id, task_instance.dag_id, session=session
-    )