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

opsgenie_closer.py

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

    def __call__(self, context) -> requests.Response:
        task_instance: TaskInstance = context['task_instance']
        if task_instance.try_number > 1:
            # create the alias
            alias = f"{ti.dag_id}.{ti.task_id}-{context['ds']}"
            
            # Post to close the alert
            hook = OpsgenieAlertHook(self.connection_id)
            return hook.run(endpoint=f'v2/alerts/{alias}/close?identifierType=alias',
                            data=json.dumps({'user': 'Airflow'}), # Who closed the alert
                            headers={'Content-Type': 'application/json',
                                     'Authorization': 'GenieKey %s' % hook._get_api_key()})