F строки в xcom pull дают неопределенную ошибку jinja в Airflow

Я пытаюсь передать строку f в параметр conf TriggerDagRunOperator. Эти задачи выполняются в цикле for, и external_dag_id изменяется на каждой итерации.

Я пробовал следующее:

previous_task_id = 'get_config_for_dag_' + external_dag_id
conf=f"{{{{ ti.xcom_pull(task_ids = {previous_task_id} , key='return_value') }}}}"

Однако я получаю ошибку jinja:

jinja2.exceptions.UndefinedError: «get_config_for_dag_bla_bla_sbx» не определен

Кто-нибудь видит ошибку в строке f, которую я строю?

Вот более полная часть кода:


dag_info = ['bla_bla_sbx']
for external_dag_id in dag_info:

    get_config_for_dag = PythonOperator(
                task_id = "get_config_for_dag_" + external_dag_id,
                python_callable=get_dict_value,
                op_kwargs = {
                    'config_dict': "{{ ti.xcom_pull(task_ids='build_dags_configuration', key='return_value') }}",
                    'current_dag_id': external_dag_id
                }
            )


    previous_task_id = 'get_config_for_dag_' + external_dag_id        
    run_dag = TriggerDagRunOperator(
                task_id = "run_dag_" + external_dag_id,
                wait_for_completion=True,
                trigger_dag_id=external_dag_id,
                pool = "dag_pool",
                conf=f"{{{{ ti.xcom_pull(task_ids = {previous_task_id} , key='return_value') }}}}",
                )


Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
1
0
74
2
Перейти к ответу Данный вопрос помечен как решенный

Ответы 2

Попробуйте добавить цитату в task_ids : '{previous_task_id}'

f"{{{{ ti.xcom_pull(task_ids='{previous_task_id}' , key='return_value') }}}}"

Я думаю, что это было связано с кавычками, но по какой-то причине это не сработало.

Fernando Garcia Dorador 05.04.2023 10:15
Ответ принят как подходящий

В итоге получилось так:

f"{{{{ ti.xcom_pull(task_ids='get_config_for_dag_{external_dag_id}' , key='return_value') }}}}"

Другие вопросы по теме