In my Airflow DAG I needed to set render_template_as_native_obj=True
to be able to customize the input value in one task. As far as I know is not possible to apply the flag only to the desired tasks, so this will affect all the tasks in the DAG.
My problem comes with env_vars of a Kubernetes Pod Operator. This is the task code, which was working before applying the flag:
download_missing_files = KubernetesPodOperator(
task_id=f"download_missing_files_{file_to_load['task_alias']}",
image="....",
namespace='dbt',
service_account_name='dbt-worker',
name="googleads",
get_logs=True,
config_file="/home/airflow/composer_kube_config",
kubernetes_conn_id='kubernetes_default',
log_events_on_failure=True,
env_vars={
"missing_files": f"{list_missing_files.output['missing_files']}",
"destination_bucket": f"{file_to_load['destination_bucket']}",
"google_cred": f"{file_to_load['source_conn_id']}",
"api_version": f"{file_to_load['g_ads_api_version']}",
"backwards_days": "{{params.backwards_days}}",
"data_owner_sa": f"{DATA_OWNER_SA}"
},
container_resources=k8s_models.V1ResourceRequirements(
requests={"cpu": "1000m", "memory": "4G", "ephemeral-storage": "2G"},
limits={"cpu": "1000m", "memory": "4G", "ephemeral-storage": "2G"},
),
do_xcom_push=True,
)
Afterwards I tried many different versions modifying the way env_var is created, but it keeps failing. This was one of the versions:
env_vars={
"missing_files": f"{{{{ ti.xcom_pull(key='manual_files_load_list', task_ids='list_missing_files_{file_to_load['task_alias']}) }}}}",
"destination_bucket": f"{{{file_to_load['destination_bucket']}}}",
"google_cred": f"{{{file_to_load['source_conn_id']}}}",
"api_version": f"{{{file_to_load['g_ads_api_version']}}}",
"backwards_days": "{{params.backwards_days}}",
"data_owner_sa": f"{{{DATA_OWNER_SA}}}"
},
The error in this case:
jinja2.exceptions.TemplateSyntaxError: expected token 'end of print statement', got ':'
Does anyone understand how does it works? Thanks in advance.