Support @task.bash with Task SDK

@task.bash code currently requires DB interaction as below, we need to remove that and use Task SDK somehow

airflow/providers/src/airflow/providers/standard/operators/bash.py

Lines 198 to 222 in f007929

    
               # TODO: This should be replaced with Task SDK API call 
        
               @staticmethod 
        
               @provide_session 
        
               def refresh_bash_command(ti: TaskInstance, session: SASession = NEW_SESSION) -> None: 
        
                   """ 
        
                   Rewrite the underlying rendered bash_command value 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 bash_command 
        
                   value. 
        
                   :meta private: 
        
                   """ 
        
                   from airflow.models.renderedtifields import RenderedTaskInstanceFields 
        
                   """Update rendered task instance fields for cases where runtime evaluated, not templated.""" 
        
                   # Note: Need lazy import to break the partly loaded class loop 
        
                   from airflow.models.taskinstance import TaskInstance 
        
                   # If called via remote API the DAG needs to be re-loaded 
        
                   TaskInstance.ensure_dag(ti, session=session) 
        
                   rtif = RenderedTaskInstanceFields(ti) 
        
                   RenderedTaskInstanceFields.write(rtif, session=session) 
        
                   RenderedTaskInstanceFields.delete_old_records(ti.task_id, ti.dag_id, session=session)

Support `@task.bash` with Task SDK

Navigation Menu

Repository navigation

Support `@task.bash` with Task SDK #45639

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	# TODO: This should be replaced with Task SDK API call
	@staticmethod
	@provide_session
	def refresh_bash_command(ti: TaskInstance, session: SASession = NEW_SESSION) -> None:
	"""
	Rewrite the underlying rendered bash_command value 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 bash_command
	value.

	:meta private:
	"""
	from airflow.models.renderedtifields import RenderedTaskInstanceFields

	"""Update rendered task instance fields for cases where runtime evaluated, not templated."""
	# Note: Need lazy import to break the partly loaded class loop
	from airflow.models.taskinstance import TaskInstance

	# If called via remote API the DAG needs to be re-loaded
	TaskInstance.ensure_dag(ti, session=session)

	rtif = RenderedTaskInstanceFields(ti)
	RenderedTaskInstanceFields.write(rtif, session=session)
	RenderedTaskInstanceFields.delete_old_records(ti.task_id, ti.dag_id, session=session)

Support @task.bash with Task SDK #45639

Description

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Support `@task.bash` with Task SDK #45639