HTTP Notifier implementation

[seanghaeli]

Implementation and unit test for HTTP Notifier. Here's the Dag I used to test that it works end-to-end

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.http.notifications import HttpNotifier

http_notifier = HttpNotifier(
    http_conn_id="local_server_conn",
    endpoint="",
    method="POST",
    json={
        "dag_id": "{{ dag.dag_id }}",
        "task_id": "{{ task.task_id }}",
        "execution_date": "{{ ds }}",
        "status": "success"
    }
)

dag = DAG(
    "test_http_notifier",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
    on_success_callback=http_notifier,
)

test_task = BashOperator(
    task_id="test_task",
    bash_command="echo 'Testing HTTP notifier'",
    dag=dag,
)

I add a connection within airflow using:

airflow connections add local_server_conn \ 
--conn-type http \ 
--conn-host host.docker.internal \ 
--conn-port 8081 \ 
--conn-schema http

And spin up a local API server via python:

python3 -c "
import http.server
import socketserver
import json
from datetime import datetime

class Handler(http.server.SimpleHTTPRequestHandler):
    def do_POST(self):
        print(f'[{datetime.now().strftime(\"%H:%M:%S\")}] 🚁 AIRFLOW NOTIFICATION!')
        
        length = int(self.headers.get('Content-Length', 0))
        data = self.rfile.read(length).decode('utf-8')
        
        try:
            json_data = json.loads(data)
            print(f'📋 Airflow sent:')
            print(json.dumps(json_data, indent=2))
        except:
            print(f'📝 Raw data: {data}')
        
        print('=' * 60)
        
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(b'{\"status\": \"notification received!\"}')

print('AIRFLOW NOTIFICATION SERVER READY')
print('Listening on http://localhost:8081')
print('=' * 60)
socketserver.TCPServer(('', 8081), Handler).serve_forever()
"

Then I run the above Dag, and on the CLI with the python API-server running I can see the contents of the POST request that is triggered when the Dag completes successfully.

[eladkal]

Notifiers needs to register in provider.yaml
similar to

airflow/providers/amazon/provider.yaml

Lines 908 to 911 in 696fd73

	notifications:
	- airflow.providers.amazon.aws.notifications.chime.ChimeNotifier
	- airflow.providers.amazon.aws.notifications.sns.SnsNotifier
	- airflow.providers.amazon.aws.notifications.sqs.SqsNotifier

I guess we are missing verification that any new notifier has corresponded entry in the provider.yaml

[seanghaeli]

Bad rebase, please ignore if you were pinged.

[jscheffl]

I like this! No problem being pinged.

As you have in the PR description an example, can you add this example to the codebase as well and a reference into docs so that people can find it? Will be useful for people to know other than taking a look to the source.