Source code for airflow.contrib.operators.jira_operator

# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from airflow.contrib.hooks.jira_hook import JIRAError
from airflow.contrib.hooks.jira_hook import JiraHook
from airflow.exceptions import AirflowException
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

[docs]class JiraOperator(BaseOperator): """ JiraOperator to interact and perform action on Jira issue tracking system. This operator is designed to use Jira Python SDK: :param jira_conn_id: reference to a pre-defined Jira Connection :type jira_conn_id: str :param jira_method: method name from Jira Python SDK to be called :type jira_method: str :param jira_method_args: required method parameters for the jira_method. (templated) :type jira_method_args: dict :param result_processor: function to further process the response from Jira :type result_processor: function :param get_jira_resource_method: function or operator to get jira resource on which the provided jira_method will be executed :type get_jira_resource_method: function """
[docs] template_fields = ("jira_method_args",)
@apply_defaults def __init__(self, jira_conn_id='jira_default', jira_method=None, jira_method_args=None, result_processor=None, get_jira_resource_method=None, *args, **kwargs): super(JiraOperator, self).__init__(*args, **kwargs) self.jira_conn_id = jira_conn_id self.method_name = jira_method self.jira_method_args = jira_method_args self.result_processor = result_processor self.get_jira_resource_method = get_jira_resource_method
[docs] def execute(self, context): try: if self.get_jira_resource_method is not None: # if get_jira_resource_method is provided, jira_method will be executed on # resource returned by executing the get_jira_resource_method. # This makes all the provided methods of JIRA sdk accessible and usable # directly at the JiraOperator without additional wrappers. # ref: if isinstance(self.get_jira_resource_method, JiraOperator): resource = self.get_jira_resource_method.execute(**context) else: resource = self.get_jira_resource_method(**context) else: # Default method execution is on the top level jira client resource hook = JiraHook(jira_conn_id=self.jira_conn_id) resource = hook.client # Current Jira-Python SDK (1.0.7) has issue with pickling the jira response. # ex: self.xcom_push(context, key='operator_response', value=jira_response) # This could potentially throw error if jira_result is not picklable jira_result = getattr(resource, self.method_name)(**self.jira_method_args) if self.result_processor: return self.result_processor(context, jira_result) return jira_result except JIRAError as jira_error: raise AirflowException("Failed to execute jiraOperator, error: %s" % str(jira_error)) except Exception as e: raise AirflowException("Jira operator error: %s" % str(e))