<commit_before>import abc
import datetime
import operator
from functools import wraps
from typing import Any, Callable, FrozenSet, Generic, List, NoReturn, Optional, TypeVar

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from rest_framework.mixins import UpdateModelMixin
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.serializers import BaseSerializer, ModelSerializer
from rest_framework.viewsets import GenericViewSet
from typing_extensions import Final, final

from bananas.admin.api.schemas.yasg import BananasSwaggerSchema
from bananas.models import TimeStampedModel

from . import errors
from .utils import HeaderError, parse_header_datetime, parse_header_etags

__all__ = (
    "Fence",
    "FencedUpdateModelMixin",
    "header_date_parser",
    "parse_date_modified",
    "allow_if_unmodified_since",
    "header_etag_parser",
    "allow_if_match",
)


TokenType = TypeVar("TokenType")
InstanceType = TypeVar("InstanceType")


@final
class Fence(abc.ABC, Generic[InstanceType, TokenType]):
    def __init__(
        self,
        get_token: Callable[[Request], TokenType],
        compare: Callable[[TokenType, TokenType], bool],
        get_version: Callable[[InstanceType], Optional[TokenType]],
        openapi_parameter: openapi.Parameter,
        rejection: Exception = errors.PreconditionFailed(
            "The resource does not fulfill the given preconditions"
        ),
    ) -> None:
        self._get_token: Final = get_token
        self._compare: Final = compare
        self._get_version: Final = get_version
        self._rejection: Final = rejection
        self.openapi_parameter: Final = openapi_parameter

    def check(self, request: Request, instance: InstanceType) -> bool:
        version = self._get_version(instance)
        if version is None:
            # We might want to expose control of this behavior. For if-unmodified-since
            # it makes sense to return true here, but it might not for
            # if-modified-since other conditionals.
            return True
        return self._compare(version, self._get_token(request))

    def reject(self) -> NoReturn:
        raise self._rejection

    def validate(self, request: Request, instance: InstanceType) -> None:
        if not self.check(request, instance):
            self.reject()


class FenceAwareSwaggerAutoSchema(BananasSwaggerSchema):
    update_methods = ("PUT", "PATCH")

    def add_manual_parameters(
        self, parameters: List[openapi.Parameter]
    ) -> List[openapi.Parameter]:
        parameters = super().add_manual_parameters(parameters)
        if (
            isinstance(self.view, FencedUpdateModelMixin)
            and self.method in self.update_methods
        ):
            return parameters + [self.view.fence.openapi_parameter]
        return parameters


class FencedUpdateModelMixin(UpdateModelMixin, abc.ABC):
    @property
    @abc.abstractmethod
    def fence(self) -> Fence:
        ...

    def perform_update(self, serializer: BaseSerializer) -> None:
        # mypy's advanced self types don't work with super calls so we use an assertion
        # here instead.
        assert isinstance(self, GenericViewSet)
        assert isinstance(serializer, ModelSerializer)
        self.fence.validate(self.request, serializer.instance)
        super().perform_update(serializer)

    @swagger_auto_schema(auto_schema=FenceAwareSwaggerAutoSchema)
    def update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
        return super().update(request, *args, **kwargs)

    @swagger_auto_schema(auto_schema=FenceAwareSwaggerAutoSchema)
    def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
        return super().partial_update(request, *args, **kwargs)


def header_date_parser(header: str) -> Callable[[Request], datetime.datetime]:
    def parse(request: Request) -> datetime.datetime:
        try:
            return parse_header_datetime(request, header)
        except HeaderError as e:
            raise e.as_api_error()

    return parse


def parse_date_modified(instance: TimeStampedModel) -> Optional[datetime.datetime]:<commit_msg>Fix broken self type (#62)<commit_after>import abc
import datetime
import operator
from functools import wraps
from typing import Any, Callable, FrozenSet, Generic, List, NoReturn, Optional, TypeVar

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.db.models import QuerySet
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from rest_framework.mixins import UpdateModelMixin
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.serializers import BaseSerializer, ModelSerializer
from rest_framework.viewsets import GenericViewSet
from typing_extensions import Final, final

from bananas.admin.api.schemas.yasg import BananasSwaggerSchema
from bananas.models import TimeStampedModel

from . import errors
from .utils import HeaderError, parse_header_datetime, parse_header_etags

__all__ = (
    "Fence",
    "FencedUpdateModelMixin",
    "header_date_parser",
    "parse_date_modified",
    "allow_if_unmodified_since",
    "header_etag_parser",
    "allow_if_match",
)


TokenType = TypeVar("TokenType")
InstanceType = TypeVar("InstanceType")


@final
class Fence(abc.ABC, Generic[InstanceType, TokenType]):
    def __init__(
        self,
        get_token: Callable[[Request], TokenType],
        compare: Callable[[TokenType, TokenType], bool],
        get_version: Callable[[InstanceType], Optional[TokenType]],
        openapi_parameter: openapi.Parameter,
        rejection: Exception = errors.PreconditionFailed(
            "The resource does not fulfill the given preconditions"
        ),
    ) -> None:
        self._get_token: Final = get_token
        self._compare: Final = compare
        self._get_version: Final = get_version
        self._rejection: Final = rejection
        self.openapi_parameter: Final = openapi_parameter

    def check(self, request: Request, instance: InstanceType) -> bool:
        version = self._get_version(instance)
        if version is None:
            # We might want to expose control of this behavior. For if-unmodified-since
            # it makes sense to return true here, but it might not for
            # if-modified-since other conditionals.
            return True
        return self._compare(version, self._get_token(request))

    def reject(self) -> NoReturn:
        raise self._rejection

    def validate(self, request: Request, instance: InstanceType) -> None:
        if not self.check(request, instance):
            self.reject()


class FenceAwareSwaggerAutoSchema(BananasSwaggerSchema):
    update_methods = ("PUT", "PATCH")

    def add_manual_parameters(
        self, parameters: List[openapi.Parameter]
    ) -> List[openapi.Parameter]:
        parameters = super().add_manual_parameters(parameters)
        if (
            isinstance(self.view, FencedUpdateModelMixin)
            and self.method in self.update_methods
        ):
            return parameters + [self.view.fence.openapi_parameter]
        return parameters


class FencedUpdateModelMixin(UpdateModelMixin, abc.ABC):
    @property
    @abc.abstractmethod
    def fence(self) -> Fence:
        ...

    # django-restframework uses an "advanced self-type" on self in
    # perform_update() which subtly breaks subclassing. We try to remedy this by
    # using an abstract method instead.
    # See https://github.com/typeddjango/djangorestframework-stubs/issues/132
    @abc.abstractmethod
    def get_queryset(self) -> QuerySet:
        ...

    def perform_update(self, serializer: BaseSerializer) -> None:
        # mypy's advanced self types don't work with super calls so we use an assertion
        # here instead.
        assert isinstance(self, GenericViewSet)
        assert isinstance(serializer, ModelSerializer)
        self.fence.validate(self.request, serializer.instance)
        super().perform_update(serializer)

    @swagger_auto_schema(auto_schema=FenceAwareSwaggerAutoSchema)
    def update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
        return super().update(request, *args, **kwargs)

    @swagger_auto_schema(auto_schema=FenceAwareSwaggerAutoSchema)
    def partial_update(self, request: Request, *args: Any, **kwargs: Any) -> Response:
        return super().partial_update(request, *args, **kwargs)


def header_date_parser(header: str) -> Callable[[Request], datetime.datetime]:
    def parse(request: Request) -> datetime.datetime:
        try:
            return parse_header_datetime(request, header)
        except HeaderError as e:
            raise e.as_api_error()

    return parse


def parse_date_modified(instance: TimeStampedModel) -> Optional[datetime.datetime]:
{'fraction_non_alphanumeric': 0.08516300731869594, 'fraction_numerical': 0.0005544466622310934, 'mean_word_length': 3.525338685398896, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 28, 'https://': 1, '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': '4474229', 'n_tokens_mistral': 2501, 'n_tokens_neox': 2447, 'n_words': 669}