Airbrake using API and Python Requests

test_airbrake.py

import json
import traceback
import requests
import os
from datetime import datetime

AIRBRAKE_PROJECT_ID = os.getenv("AIRBRAKE_PROJECT_ID")
AIRBRAKE_PROJECT_KEY = os.getenv("AIRBRAKE_PROJECT_KEY")


def notify_airbrake(
    error, context=None, environment="development", project_id=None, project_key=None
):
    if project_id is None or project_key is None:
        raise ValueError("Both project_id and project_key are required")

    # Get error details
    if isinstance(error, Exception):
        error_class = error.__class__.__name__
        error_message = str(error)
        backtrace = traceback.format_exc()
    else:
        error_class = "Error"
        error_message = str(error)
        backtrace = "".join(traceback.format_stack())

    # Prepare the payload according to Airbrake API v3
    payload = {
        "notifier": {
            "name": "python-airbrake-direct",
            "version": "1.0.0",
            "url": "https://github.com/airbrake/airbrake",
        },
        "errors": [
            {
                "type": error_class,
                "message": error_message,
                "backtrace": [
                    {
                        "file": line.split('"')[1],
                        "function": line.split("in ")[1].strip(),
                        "line": (
                            int(line.split(", line ")[1].split(",")[0])
                            if ", line " in line
                            else 0
                        ),
                    }
                    for line in backtrace.split("\n")
                    if line.strip() and "File " in line
                ],
            }
        ],
        "context": {
            "environment": environment,
            "notifier": {
                "name": "python-airbrake-direct",
                "version": "1.0.0",
                "url": "https://github.com/airbrake/airbrake",
            },
            "timestamp": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
            **(context or {}),
        },
    }

    # Send the request to Airbrake API
    url = f"https://api.airbrake.io/api/v3/projects/{project_id}/notices"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {project_key}",
    }

    response = requests.post(url, headers=headers, data=json.dumps(payload))
    return response.status_code, (
        response.json() if response.status_code == 201 else response.text
    )


def test_airbrake():
    try:
        # Some code that might raise an exception
        1 / 0
    except Exception as e:
        notify_airbrake(
            error=e,
            context={"user_id": "123", "component": "calculation_service"},
            environment="development",
            project_id=AIRBRAKE_PROJECT_ID,
            project_key=AIRBRAKE_PROJECT_KEY,
        )