Blogpost on OpsGenie alerting in Airflow https://medium.com/p/239ddea61d0a

opsgenie_reporter_simple.py

class OpsGenieExceptionReporter():
    def __init__(self, connection_id: str = opsgenie_default,) -> None:
        self.connection_id = connection_id

    def __call__(self, context) -> requests.Response:
        ti: TaskInstance = context["ti"]
        json = {
            "message": f"Something went terribly wrong with {ti.task_id}",
            "description": f"See more at {ti.log_url}",
            "responders": [{"name": "my-fancy-team", "type": "team"}],
            "alias": f"{ti.dag_id}.{ti.task_id}-{context['ds']}"
        }
        return OpsgenieAlertHook(self.connection_id).execute(json)