[Discord Provider] Add support to embeds in Discord Webhook and Notifier

[1cadumagalhaes]

Description

I would like to be able to send more complex discord messages, not only text formatted.

We are limited when using only the content/message field. This is an example of a notification we use today

And this is a message in a custom notifier for airbyte

To create this, we could either create the option to use an embed for the users, and let them format it; or define a template and set the fields in the notifier/webhook. This isnt complicated to do, here is the builder for the airbyte notifier

def format_airbyte_data(airbyte_data):
    """Convert Airbyte data payload (for success/failure) into a Discord embed format."""
    data = airbyte_data.get('data', {})
    if not data:
        return None
        
    # Determine status color (green for success, red for failure)
    color = 0x00FF00 if data.get('success', False) else 0xFF0000
    
    # Create the main embed
    embed = {
        "title": f"Airbyte Sync: {data.get('connection', {}).get('name', 'Unknown')}",
        "color": color,
        "fields": [
            {
                "name": "Status",
                "value": "✅ Success" if data.get('success', False) else "❌ Failed",
                "inline": True
            },
            {
                "name": "Duration",
                "value": data.get('durationFormatted', 'Unknown'),
                "inline": True
            },
            {
                "name": "Records",
                "value": f"Emitted: {data.get('recordsEmitted', 0)}\nCommitted: {data.get('recordsCommitted', 0)}",
                "inline": True
            },
            {
                "name": "Data Volume",
                "value": f"Emitted: {data.get('bytesEmittedFormatted', '0 B')}\nCommitted: {data.get('bytesCommittedFormatted', '0 B')}",
                "inline": True
            },
            {
                "name": "Source",
                "value": data.get('source', {}).get('name', 'Unknown'),
                "inline": True
            },
            {
                "name": "Destination",
                "value": data.get('destination', {}).get('name', 'Unknown'),
                "inline": True
            }
        ],
        "footer": {
            "text": f"Workspace: {data.get('workspace', {}).get('name', 'Unknown')} | Job ID: {data.get('jobId', 'Unknown')}"
        },
        "timestamp": data.get('finishedAt', datetime.utcnow().isoformat())
    }
    # Add URLs as a separate non-inline field if they exist
    urls = []
    if data.get('connection', {}).get('url'):
        urls.append(f"[View Connection]({data['connection']['url']})")
    if data.get('source', {}).get('url'):
        urls.append(f"[View Source]({data['source']['url']})")
    if data.get('destination', {}).get('url'):
        urls.append(f"[View Destination]({data['destination']['url']})")
    
    if urls:
        embed["fields"].append({
            "name": "Links",
            "value": " | ".join(urls),
            "inline": False
        })
    
    return {
        "embeds": [embed]
    }

Honestly this is quite easy and I'll probably have a version of it when I get a few minutes to work on it

Use case/motivation

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

Feel free to add it.

[1cadumagalhaes]

So, in my instance I've set the notifier to get information from the task_instance context.

I've added a few new fields, so the developer/user can use them to overwrite any of these values

        self.success = success
        self.success_color = success_color
        self.failure_color = failure_color
        self.mentions = mentions
        self.mention_everyone = mention_everyone
        self.mention_roles = mention_roles
        self.mention_channels = mention_channels
        self.application_id = application_id
        self.dag_name = None
        self.task_name = None
        self.duration = None
        self.failure_link = None
        self.success_link = None
        self.reactions = None
        self.proxy = None
        self.embeds = None
        self.description = description

And I've used the task_instance to get some of the informations I wanted to display

    def notify(self, context):
        """Send a message to a Discord channel."""
        task_instance = context.get("task_instance")
        if task_instance:
            self.dag_name = task_instance.dag_id.replace("_", "\\_")
            self.task_name = task_instance.task_id
            self.duration = str(task_instance.duration)
            self.failure_link = f"https://localhost:8080/dags/{task_instance.dag_id}/grid?task_id={task_instance.task_id}&tab=logs&dag_run_id={urlencode({'dag_run_id': task_instance.run_id.replace('+', '%2B')})}"
            self.success_link = f"https://localhost:8080/dags/{task_instance.dag_id}/grid?task_id={task_instance.task_id}&tab=logs&dag_run_id={urlencode({'dag_run_id': task_instance.run_id.replace('+', '%2B')})}"

I'm not sure if this is the best approach for the project as a whole, since what I did basically was to generate the embed in the notifier and pass it to the hook. Maybe we could have a discussion on what could be important to display in a suggested default notification from airflow, and let the users customize it if they want.

[1cadumagalhaes]

I mean, the main advantage of setting a default embed is that it would be way easier for the users to implement notifications, not having to deal with the content as a whole. Just passing the notifier on the callbacks, and calling it with the connection name.

[potiuk]

Create a PR and discuss it there.

[dondaum]

@eladkal, can you please assign this to me ?