ASU-S DOCS

Appearance

Webhooks ​

Установка в проект ​

  1. Добавить asu_s_sdk.django.webhooks в INSTALLED_APPS
  2. Создать django app webhooks (./manage.py startapp webhooks) и добавить его в INSTALLED_APPS. После создания в apps.py добавить метод def ready(self) и сделать в нём импорт from . import handlers # NOQA (пакет handlers нужно создать, далее в нём должны будут хранится все обработчики вебхуков).
  3. Накатить миграции ./manage.py migrate
  4. Удостовериться, что в проекте установлен celery и он работает.
  5. В urls.py добавить роутер path("webhooks/", include("asu_s_sdk.django.webhooks.urls")). Обычно этот urls.py расположен в project.api.internal.v1.urls.py. Если в проекте отсутствует project.api.internal.v1.urls.py его нужно создать, для примера можно ориентироваться на asu-s-core.
  6. Добавить коннектор для вебхуков в asu_s_sdk.django.webhooks.services.connectors и значение в enum asu_s_sdk.django.webhooks.enums.

Пример структуры директорий для webhooks app ​

project/
    webhooks/
        handlers/
            __init__.py
            type_of_event.py
        apps.py

Жизненный цикл вебхука ​

  1. Запись в БД данных на отправку вебхука (url, method, headers, body запроса)
  2. Отправка вебхука через HTTP
  3. Запись в БД информации об ответе от внешней системы (status_code, headers, body)
  4. Внешняя система получает вебхук
  5. Регистрация вебхука в БД внешней системы (ip_address, headers, body)
  6. Обработка вебхука внешней системой в очереди celery
  7. Обновление статуса обработки вебхука в БД

Отправка вебхуков ​

Все вебхуки представляют собой обычные преднастроенные HTTP запросы. На случай недоступности или загруженности принимающего сервера установлен timeout в 30 секунд и 5 попыток повторной отправки на случай получения ошибок [422, 500, 502, 503, 504]. Отправка вебхуков происходит с авторизацией в заголовках через внутренний токен ASU (см. asu_s_sdk.utils.rest_framework.auth.authentication.AsuInternalTokenAuthentication).

Как отправить? ​

python
from asu_s_sdk.django.webhooks.services.request import WebhookRequestService
from asu_s_sdk.django.webhooks.enums import WebhookRequestConnectorType

WebhookRequestService(
    WebhookRequestConnectorType.APPROVALS
).create(
    type="TYPE_OF_EVENT",
    data={
        "foo": "bar",
    },
)

В этом примере:

  • WebhookRequestConnectorType.APPROVALS — название ASU сервиса, куда хотим отправить (должен быть добавлен в asu_s_sdk, если его нет)
  • type — название события
  • data — любой dict, обязательно должен быть JSON сериализуемым (нельзя, чтобы значениями были datetime, uuid4, Decimal и другие)

Как отправить с помощью celery? ​

python
from asu_s_sdk.django.webhooks.tasks import send_asu_webhook_task

send_asu_webhook_task.delay("APPROVALS", "TYPE_OF_EVENT", {"foo": "bar"})

Если произошла ошибка отправки? ​

Все действия логгируются и сохраняются в базу данных в WebhookRequest, в django admin есть возможность повторно отправить вебхук.

Получение вебхуков ​

Получение вебхуков это входящие HTTP запросы в тот или иной сервис. Ранее в примере отправки мы отправляли вебхук с типом TYPE_OF_EVENT в сервис APPROVALS. Это значит, что сервис APPROVALS должен уметь обработать тело запроса для типа TYPE_OF_EVENT.

  1. В пакете webhooks.handlers мы создаём type_of_event.py
  2. Создаём обработчик этого типа
python
from asu_s_sdk.django.webhooks.services.registry import register_webhook_event_handler
from asu_s_sdk.django.webhooks.services.handlers.abstract import AbstractWebhookEventHandler

@register_webhook_event_handler("TYPE_OF_EVENT")
class TypeOfEventWebhookEventHandler(AbstractWebhookEventHandler):
    def handle(self):
        return
  • В register_webhook_event_handler мы регистрируем тип события переданный при отправке
  • В методе handle мы можем обрабатывать тело вебхука как угодно, используя self.data
  1. В webhooks.handlers.__init__.py нужно обязательно импортировать вновь созданный класс TypeOfEventWebhookEventHandler

Если произошла ошибка во время обработки? ​

Все действия логгируются и сохраняются в базу данных в WebhookEvent, в django admin есть возможность повторно перезапустить обработку вебхука (в основном потоке или через очередь celery).