models.py

from core.utils.common import load_func
from core.validators import JSONSchemaValidator
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import models
from django.utils.translation import gettext_lazy as _
from labels_manager.models import LabelLink
from projects.models import Project
from tasks.models import Annotation, Task

# from labels_manager.serializers import LabelLinkSerializer, LabelSerializer
from .serializers_for_hooks import (
    OnlyIDWebhookSerializer,
)

HEADERS_SCHEMA = {
    'type': 'object',
    'patternProperties': {
        '^[a-zA-Z0-9-_]+$': {'type': 'string'},
    },
    'maxProperties': 10,
    'additionalProperties': False,
}


class Webhook(models.Model):
    """Model of webhooks.

    If webhook has not null project field -- it's project webhook
    """

    organization = models.ForeignKey('organizations.Organization', on_delete=models.CASCADE, related_name='webhooks')

    project = models.ForeignKey(
        'projects.Project', null=True, on_delete=models.CASCADE, related_name='webhooks', default=None
    )

    url = models.URLField(_('URL of webhook'), max_length=2048, help_text=_('URL of webhook'))

    send_payload = models.BooleanField(
        _('does webhook send the payload'),
        default=True,
        help_text=('If value is False send only action'),
    )

    send_for_all_actions = models.BooleanField(
        _('Use webhook for all actions'),
        default=True,
        help_text='If value is False - used only for actions from WebhookAction',
    )

    headers = models.JSONField(
        _('request extra headers of webhook'),
        validators=[JSONSchemaValidator(HEADERS_SCHEMA)],
        default=dict,
        help_text='Key Value Json of headers',
    )

    is_active = models.BooleanField(
        _('is webhook active'),
        default=True,
        help_text=('If value is False the webhook is disabled'),
    )

    created_at = models.DateTimeField(_('created at'), auto_now_add=True, help_text=_('Creation time'), db_index=True)
    updated_at = models.DateTimeField(_('updated at'), auto_now=True, help_text=_('Last update time'), db_index=True)

    def get_actions(self):
        return WebhookAction.objects.filter(webhook=self).values_list('action', flat=True)

    def validate_actions(self, actions):
        actions_meta = [WebhookAction.ACTIONS[action] for action in actions]
        if self.project and any((meta.get('organization-only') for meta in actions_meta)):
            raise ValidationError("Project webhook can't contain organization-only action.")
        return actions

    def set_actions(self, actions):
        if not actions:
            actions = set()
        actions = set(actions)
        old_actions = set(self.get_actions())

        for new_action in list(actions - old_actions):
            WebhookAction.objects.create(webhook=self, action=new_action)

        WebhookAction.objects.filter(webhook=self, action__in=(old_actions - actions)).delete()

    def has_permission(self, user):
        user.project = self.project  # link for activity log
        return self.organization.has_user(user)

    class Meta:
        db_table = 'webhook'


class WebhookAction(models.Model):
    PROJECT_CREATED = 'PROJECT_CREATED'
    PROJECT_UPDATED = 'PROJECT_UPDATED'
    PROJECT_DELETED = 'PROJECT_DELETED'

    TASKS_CREATED = 'TASKS_CREATED'
    TASKS_DELETED = 'TASKS_DELETED'

    ANNOTATION_CREATED = 'ANNOTATION_CREATED'
    ANNOTATIONS_CREATED = 'ANNOTATIONS_CREATED'
    ANNOTATION_UPDATED = 'ANNOTATION_UPDATED'
    ANNOTATIONS_DELETED = 'ANNOTATIONS_DELETED'

    LABEL_LINK_CREATED = 'LABEL_LINK_CREATED'
    LABEL_LINK_UPDATED = 'LABEL_LINK_UPDATED'
    LABEL_LINK_DELETED = 'LABEL_LINK_DELETED'

    ACTIONS = {
        PROJECT_CREATED: {
            'name': _('Project created'),
            'description': _(''),
            'key': 'project',
            'many': False,
            'model': Project,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['project']),
            'organization-only': True,
        },
        PROJECT_UPDATED: {
            'name': _('Project updated'),
            'description': _(''),
            'key': 'project',
            'many': False,
            'model': Project,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['project']),
            'project-field': '__self__',
        },
        PROJECT_DELETED: {
            'name': _('Project deleted'),
            'description': _(''),
            'key': 'project',
            'many': False,
            'model': Project,
            'serializer': OnlyIDWebhookSerializer,
            'organization-only': True,
        },
        TASKS_CREATED: {
            'name': _('Task created'),
            'description': _(''),
            'key': 'tasks',
            'many': True,
            'model': Task,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['task']),
            'project-field': 'project',
        },
        TASKS_DELETED: {
            'name': _('Task deleted'),
            'description': _(''),
            'key': 'tasks',
            'many': True,
            'model': Task,
            'serializer': OnlyIDWebhookSerializer,
            'project-field': 'project',
        },
        ANNOTATION_CREATED: {
            'name': _('Annotation created'),
            'description': _(''),
            'key': 'annotation',
            'many': False,
            'model': Annotation,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['annotation']),
            'project-field': 'project',
            'nested-fields': {
                'task': {
                    'serializer': load_func(settings.WEBHOOK_SERIALIZERS['task']),
                    'many': False,
                    'field': 'task',
                },
            },
        },
        ANNOTATIONS_CREATED: {
            'name': _('Annotations created'),
            'description': _(''),
            'key': 'annotation',
            'many': True,
            'model': Annotation,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['annotation']),
            'project-field': 'project',
            'nested-fields': {
                'task': {
                    'serializer': load_func(settings.WEBHOOK_SERIALIZERS['task']),
                    'many': True,
                    'field': 'task',
                },
            },
        },
        ANNOTATION_UPDATED: {
            'name': _('Annotation updated'),
            'description': _(''),
            'key': 'annotation',
            'many': False,
            'model': Annotation,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['annotation']),
            'project-field': 'project',
            'nested-fields': {
                'task': {
                    'serializer': load_func(settings.WEBHOOK_SERIALIZERS['task']),
                    'many': False,
                    'field': 'task',
                },
            },
        },
        ANNOTATIONS_DELETED: {
            'name': _('Annotation deleted'),
            'description': _(''),
            'key': 'annotations',
            'many': True,
            'model': Annotation,
            'serializer': OnlyIDWebhookSerializer,
            'project-field': 'project',
        },
        LABEL_LINK_CREATED: {
            'name': _('Label link created'),
            'description': _(''),
            'key': 'label_link',
            'many': True,
            'model': LabelLink,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['label_link']),
            'project-field': 'project',
        },
        LABEL_LINK_UPDATED: {
            'name': _('Label link updated'),
            'description': _(''),
            'key': 'label_link',
            'many': False,
            'model': LabelLink,
            'serializer': load_func(settings.WEBHOOK_SERIALIZERS['label_link']),
            'project-field': 'project',
            'nested-fields': {
                'label': {
                    'many': False,
                    'field': 'label',
                    'serializer': load_func(settings.WEBHOOK_SERIALIZERS['label']),
                },
            },
        },
        LABEL_LINK_DELETED: {
            'name': _('Label link deleted'),
            'description': _(''),
            'key': 'label_link',
            'many': False,
            'model': LabelLink,
            'serializer': OnlyIDWebhookSerializer,
            'project-field': 'project',
        },
    }

    webhook = models.ForeignKey(Webhook, on_delete=models.CASCADE, related_name='actions')

    action = models.CharField(
        _('action of webhook'),
        choices=[[key, value['name']] for key, value in ACTIONS.items()],
        max_length=128,
        db_index=True,
        help_text=_('Action value'),
    )

    class Meta:
        db_table = 'webhook_action'
        unique_together = [['webhook', 'action']]