Function trigger_feature_state_change_webhooks

api/features/tasks.py:23–64

(  # type: ignore[no-untyped-def]
    instance: FeatureState, event_type: WebhookEventType = WebhookEventType.FLAG_UPDATED
)

def trigger_feature_state_change_webhooks(  # type: ignore[no-untyped-def]
    instance: FeatureState, event_type: WebhookEventType = WebhookEventType.FLAG_UPDATED
):
    assert event_type in [WebhookEventType.FLAG_UPDATED, WebhookEventType.FLAG_DELETED]

    history_instance = instance.history.first()
    timestamp = (
        history_instance.history_date.strftime(WEBHOOK_DATETIME_FORMAT)
        if history_instance and history_instance.history_date
        else ""
    )
    changed_by = (
        history_instance.history_user.email
        if history_instance and history_instance.history_user
        else (
            history_instance.master_api_key.name
            if history_instance and history_instance.master_api_key
            else ""
        )
    )
    new_state = (
        None
        if event_type == WebhookEventType.FLAG_DELETED
        else _get_feature_state_webhook_data(instance)
    )
    data = {"new_state": new_state, "changed_by": changed_by, "timestamp": timestamp}
    previous_state = _get_previous_state(instance, history_instance, event_type)

    if previous_state:
        data.update(previous_state=previous_state)

    call_environment_webhooks.delay(
        args=(instance.environment.id, data, event_type.value)  # type: ignore[union-attr]
    )

    call_organisation_webhooks.delay(
        args=(
            instance.environment.project.organisation.id,  # type: ignore[union-attr]
            data,
            event_type.value,
        )
    )


def _get_previous_state(