Source code for airflow.contrib.operators.opsgenie_alert_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
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
from airflow.contrib.hooks.opsgenie_alert_hook import OpsgenieAlertHook
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults



[docs]class OpsgenieAlertOperator(BaseOperator):
    """
    This operator allows you to post alerts to Opsgenie.
    Accepts a connection that has an Opsgenie API key as the connection's password.
    This operator sets the domain to conn_id.host, and if not set will default
    to ``https://api.opsgenie.com``.

    Each Opsgenie API key can be pre-configured to a team integration.
    You can override these defaults in this operator.

    :param opsgenie_conn_id: The name of the Opsgenie connection to use
    :type opsgenie_conn_id: str
    :param message: The Message of the Opsgenie alert (templated)
    :type message: str
    :param alias: Client-defined identifier of the alert (templated)
    :type alias: str
    :param description: Description field of the alert (templated)
    :type description: str
    :param responders: Teams, users, escalations and schedules that
        the alert will be routed to send notifications.
    :type responders: list[dict]
    :param visibleTo: Teams and users that the alert will become visible
        to without sending any notification.
    :type visibleTo: list[dict]
    :param actions: Custom actions that will be available for the alert.
    :type actions: list[str]
    :param tags: Tags of the alert.
    :type tags: list[str]
    :param details: Map of key-value pairs to use as custom properties of the alert.
    :type details: dict
    :param entity: Entity field of the alert that is
        generally used to specify which domain alert is related to. (templated)
    :type entity: str
    :param source: Source field of the alert. Default value is
        IP address of the incoming request.
    :type source: str
    :param priority: Priority level of the alert. Default value is P3. (templated)
    :type priority: str
    :param user: Display name of the request owner.
    :type user: str
    :param note: Additional note that will be added while creating the alert. (templated)
    :type note: str
    """

[docs]    template_fields = ('message', 'alias', 'description', 'entity', 'priority', 'note')


    @apply_defaults
    def __init__(self,
                 message,
                 opsgenie_conn_id='opsgenie_default',
                 alias=None,
                 description=None,
                 responders=None,
                 visibleTo=None,
                 actions=None,
                 tags=None,
                 details=None,
                 entity=None,
                 source=None,
                 priority=None,
                 user=None,
                 note=None,
                 *args,
                 **kwargs
                 ):
        super(OpsgenieAlertOperator, self).__init__(*args, **kwargs)

        self.message = message
        self.opsgenie_conn_id = opsgenie_conn_id
        self.alias = alias
        self.description = description
        self.responders = responders
        self.visibleTo = visibleTo
        self.actions = actions
        self.tags = tags
        self.details = details
        self.entity = entity
        self.source = source
        self.priority = priority
        self.user = user
        self.note = note
        self.hook = None


[docs]    def _build_opsgenie_payload(self):
        """
        Construct the Opsgenie JSON payload. All relevant parameters are combined here
        to a valid Opsgenie JSON payload.

        :return: Opsgenie payload (dict) to send
        """
        payload = {}

        for key in [
            "message", "alias", "description", "responders",
            "visibleTo", "actions", "tags", "details", "entity",
            "source", "priority", "user", "note"
        ]:
            val = getattr(self, key)
            if val:
                payload[key] = val
        return payload



[docs]    def execute(self, context):
        """
        Call the OpsgenieAlertHook to post message
        """
        self.hook = OpsgenieAlertHook(self.opsgenie_conn_id)
        self.hook.execute(self._build_opsgenie_payload())

Was this entry helpful?