时间戳

首页 > 解决方案 > Airflow 2.1 JiraTicketSensor DuplicateTaskId 如果指定了method_name

airflow - Airflow 2.1 JiraTicketSensor DuplicateTaskId 如果指定了method_name

问题描述

我一直在尝试使用 Jira Operator 从 Airflow 创建票证并使用 JiraTicketSensor 检查票证是否已解决。创建任务工作正常，但我无法让传感器工作。

如果我不提供 method_name，我会收到一个错误，如果我将它提供为 None，我会收到一个错误，指出任务 ID 已添加到 DAG。

我在 2.0.0 上运行 Airflow 2.1 和 apache-airflow-providers-jira 包。

任何有关如何解决此问题的帮助将不胜感激。您可以在下面找到 DAG 的代码：

from airflow.providers.jira.operators.jira import JiraOperator
from airflow.providers.jira.sensors.jira import JiraTicketSensor
from airflow.utils.dates import days_ago
from airflow import DAG

issue_dict = {
    'project': {'id': 101},
    'summary': 'Confirmation required (TEST)',
    'description': 'Test for creating tickets from airflow',
    'issuetype': {'name': 'Task'},
}

extract_issue_key = lambda _, issue: issue.id

default_args = {
    'start_date': days_ago(1)
}

with DAG(
    dag_id='create_jira_ticket', 
    default_args=default_args, 
    schedule_interval='@daily',
    catchup=False) as dag:

    create_jira_issue = JiraOperator(
        task_id='get_human_approval',
        jira_conn_id='kg_jira',
        jira_method='create_issue',
        jira_method_args=issue_dict,
        result_processor=extract_issue_key
    )

    sensor = JiraTicketSensor(
        task_id='check_if_approved',
        jira_conn_id='kg_jira',
        ticket_id="{{ task_instance.xcom.pull('get_human_approval', key='return_value') }}", 
        field='status',
        expected_value='Done'
    )

    create_jira_issue >> sensor

标签： airflowjira

解决方案

推荐阅读