<commit_before>from typing import Any, Dict, Optional

from celery.utils.log import get_task_logger
from django.core.exceptions import ValidationError
from django.conf import settings
from django.contrib.sites.models import Site
from django.http import HttpRequest
from django.utils import timezone
from django.utils.functional import SimpleLazyObject
from graphql import GraphQLDocument, get_default_backend, parse
from graphql.error import GraphQLError, GraphQLSyntaxError
from graphql.language.ast import FragmentDefinition, OperationDefinition
from promise import Promise

from ...app.models import App
from ...webhook.error_codes import WebhookErrorCode
from ...core.exceptions import PermissionDenied
from ...discount.utils import fetch_discounts
from ...plugins.manager import PluginsManager
from ...settings import get_host
from ..utils import format_error

logger = get_task_logger(__name__)


def validate_subscription_query(query: str) -> bool:
    from ..api import schema

    graphql_backend = get_default_backend()
    try:
        document = graphql_backend.document_from_string(schema, query)
    except (ValueError, GraphQLSyntaxError):
        return False
    if not check_document_is_single_subscription(document):
        return False
    return True


def validate_query(query):
    if not query:
        return
    is_valid = validate_subscription_query(query)
    if not is_valid:
        raise ValidationError(
            {
                "query": ValidationError(
                    "Subscription query is not valid",
                    code=WebhookErrorCode.INVALID.value,
                )<commit_msg>isort<commit_after>from typing import Any, Dict, Optional

from celery.utils.log import get_task_logger
from django.conf import settings
from django.contrib.sites.models import Site
from django.core.exceptions import ValidationError
from django.http import HttpRequest
from django.utils import timezone
from django.utils.functional import SimpleLazyObject
from graphql import GraphQLDocument, get_default_backend, parse
from graphql.error import GraphQLError, GraphQLSyntaxError
from graphql.language.ast import FragmentDefinition, OperationDefinition
from promise import Promise

from ...app.models import App
from ...core.exceptions import PermissionDenied
from ...discount.utils import fetch_discounts
from ...plugins.manager import PluginsManager
from ...settings import get_host
from ...webhook.error_codes import WebhookErrorCode
from ..utils import format_error

logger = get_task_logger(__name__)


def validate_subscription_query(query: str) -> bool:
    from ..api import schema

    graphql_backend = get_default_backend()
    try:
        document = graphql_backend.document_from_string(schema, query)
    except (ValueError, GraphQLSyntaxError):
        return False
    if not check_document_is_single_subscription(document):
        return False
    return True


def validate_query(query):
    if not query:
        return
    is_valid = validate_subscription_query(query)
    if not is_valid:
        raise ValidationError(
            {
                "query": ValidationError(
                    "Subscription query is not valid",
                    code=WebhookErrorCode.INVALID.value,
                )
{'fraction_non_alphanumeric': 0.06012269938650307, 'fraction_numerical': 0.0, 'mean_word_length': 3.6386913229018494, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5387388', 'n_tokens_mistral': 840, 'n_tokens_neox': 818, 'n_words': 262}