0008_auto_20230712_1613.py

# Generated by Django 3.2.19 on 2023-07-12 16:13
import logging

from django.core.exceptions import ObjectDoesNotExist
from django.db import migrations

from apps.alerts.models import EscalationPolicy
from apps.webhooks.models import Webhook

LEGACY_SUFFIX = " (Legacy)"

logger = logging.getLogger(__name__)


def convert_custom_button_to_webhook(apps, schema_editor):
    CustomButton = apps.get_model("alerts", "CustomButton")
    Webhooks = apps.get_model("webhooks", "Webhook")
    EscalationPolicies = apps.get_model("alerts", "EscalationPolicy")

    # CustomButtonManager isn't used
    for cb in CustomButton.objects.filter(deleted_at=None):
        webhook, _ = Webhooks.objects.get_or_create(
            organization=cb.organization,
            team=cb.team,
            name=cb.name[:100 - len(LEGACY_SUFFIX)] + LEGACY_SUFFIX,
            is_legacy=True,
            defaults=dict(
                created_at=cb.created_at,
                url=cb.webhook,
                username=cb.user,
                password=cb.password,
                authorization_header=cb.authorization_header,
                trigger_type=Webhook.TRIGGER_ESCALATION_STEP,
                forward_all=cb.forward_whole_payload,
                data=cb.data,
            )
        )
        # migrate related escalation policies
        EscalationPolicies.objects.filter(
            step=EscalationPolicy._DEPRECATED_STEP_TRIGGER_CUSTOM_BUTTON,
            custom_button_trigger=cb,
        ).update(
            step=EscalationPolicy.STEP_TRIGGER_CUSTOM_WEBHOOK,
            custom_webhook=webhook,
        )


def undo_custom_button_to_webhook(apps, schema_editor):
    CustomButton = apps.get_model("alerts", "CustomButton")
    Webhooks = apps.get_model("webhooks", "Webhook")
    EscalationPolicies = apps.get_model("alerts", "EscalationPolicy")

    for webhook in Webhooks.objects.filter(is_legacy=True):
        try:
            cb = CustomButton.objects.get(name__startswith=webhook.name.removesuffix(LEGACY_SUFFIX), team=webhook.team, organization=webhook.organization)
        except ObjectDoesNotExist:
            logger.warning(f"Did not find matching custom button to revert {webhook.name} {webhook.organization.stack_slug}, skipping")
            continue

        EscalationPolicies.objects.filter(
            step=EscalationPolicy.STEP_TRIGGER_CUSTOM_WEBHOOK,
            custom_webhook=webhook,
        ).update(
            step=EscalationPolicy._DEPRECATED_STEP_TRIGGER_CUSTOM_BUTTON,
            custom_button_trigger=cb,
            custom_webhook=None,
        )
        webhook.delete()


class Migration(migrations.Migration):

    dependencies = [
        ('webhooks', '0007_webhookresponse_event_data'),
    ]

    operations = [
        migrations.RunPython(convert_custom_button_to_webhook, undo_custom_button_to_webhook)
    ]