<commit_before>"""
the sanic extension
"""

from functools import wraps, partial
import logging
import six
import sys
import inspect

from limits.errors import ConfigurationError
from limits.storage import storage_from_string
from limits.strategies import STRATEGIES
from limits.util import parse_many
from sanic.blueprints import Blueprint

from .errors import RateLimitExceeded
from .util import get_remote_address


class C:
    ENABLED = "RATELIMIT_ENABLED"
    STORAGE_URL = "RATELIMIT_STORAGE_URL"
    STORAGE_OPTIONS = "RATELIMIT_STORAGE_OPTIONS"
    STRATEGY = "RATELIMIT_STRATEGY"
    GLOBAL_LIMITS = "RATELIMIT_GLOBAL"
    SWALLOW_ERRORS = "RATELIMIT_SWALLOW_ERRORS"


class ExtLimit(object):
    """
    simple wrapper to encapsulate limits and their context
    """

    def __init__(self, limit, key_func, scope, per_method, methods, error_message,
                 exempt_when):
        self._limit = limit
        self.key_func = key_func
        self._scope = scope
        self.per_method = per_method
        self.methods = methods and [m.lower() for m in methods] or methods
        self.error_message = error_message
        self.exempt_when = exempt_when

    @property
    def limit(self):
        return self._limit() if callable(self._limit) else self._limit

    @property
    def scope(self):
        return self._scope

    @property
    def is_exempt(self):
        """Check if the limit is exempt."""
        return self.exempt_when and self.exempt_when()


class Limiter(object):
    """
    :param app: :class:`sanic.Sanic` instance to initialize the extension with.
    :param list global_limits: a variable list of strings denoting global
     limits to apply to all routes. :ref:`ratelimit-string` for  more details.
    :param function key_func: a callable that returns the key to rate limit by.
    :param str strategy: the strategy to use. refer to :ref:`ratelimit-strategy`
    :param str storage_uri: the storage location. refer to :ref:`ratelimit-conf`
    :param dict storage_options: kwargs to pass to the storage implementation upon instantiation.
    :param bool swallow_errors: whether to swallow errors when hitting a rate limit.
     An exception will still be logged. default ``False``
    """

    def __init__(self, app=None
                 , key_func=None
                 , global_limits=[]
                 , strategy=None
                 , storage_uri=None
                 , storage_options={}
                 , swallow_errors=False
                 ):
        self.app = app
        self.logger = logging.getLogger("sanic-limiter")

        self.enabled = True
        self._global_limits = []
        self._exempt_routes = set()
        self._request_filters = []
        self._strategy = strategy
        self._storage_uri = storage_uri
        self._storage_options = storage_options
        self._swallow_errors = swallow_errors
        self._key_func = key_func or get_remote_address
        for limit in global_limits:
            self._global_limits.extend(
                [
                    ExtLimit(
                        limit, self._key_func, None, False, None, None, None
                    ) for limit in parse_many(limit)
                    ]
            )
        self._route_limits = {}
        self._dynamic_route_limits = {}
        self._blueprint_dynamic_limits = {}
        self._blueprint_limits = {}
        self._storage = None
        self._limiter = None
        self._storage_dead = False

        class BlackHoleHandler(logging.StreamHandler):
            def emit(*_):
                return

        self.logger.addHandler(BlackHoleHandler())
        if app:
            self.init_app(app)

    def init_app(self, app):
        """
        :param app: :class:`sanic.Sanic` instance to rate limit.
        """
        self.enabled = app.config.setdefault(C.ENABLED, True)
        self._swallow_errors = app.config.setdefault(
            C.SWALLOW_ERRORS, self._swallow_errors
        )
        self._storage_options.update(
            app.config.get(C.STORAGE_OPTIONS, {})
        )
        self._storage = storage_from_string(
            self._storage_uri
            or app.config.setdefault(C.STORAGE_URL, 'memory://'),
            **self._storage_options
        )
        strategy = (
            self._strategy
            or app.config.setdefault(C.STRATEGY, 'fixed-window')
        )
        if strategy not in STRATEGIES:
            raise ConfigurationError("Invalid rate limiting strategy %s" % strategy)
        self._limiter = STRATEGIES[strategy](self._storage)

        conf_limits = app.config.get(C.GLOBAL_LIMITS, None)
        if not self._global_limits and conf_limits:
            self._global_limits = [
                ExtLimit(
                    limit, self._key_func, None, False, None, None, None
                ) for limit in parse_many(conf_limits)
                ]
        app.request_middleware.append(self.__check_request_limit)

    @property
    def limiter(self):
        return self._limiter

    def __check_request_limit(self, request):
        endpoint = request.url or ""
        view_handler = self.app.router.routes_static.get(endpoint, None)
        if view_handler is None:
            return
        view_func = view_handler.handler
        view_bpname = view_func.__dict__.get('__blueprintname__', None)<commit_msg>adapt to sanic@0.5.4<commit_after>"""
the sanic extension
"""

from functools import wraps, partial
import logging
import six
import sys
import inspect

from limits.errors import ConfigurationError
from limits.storage import storage_from_string
from limits.strategies import STRATEGIES
from limits.util import parse_many
from sanic.blueprints import Blueprint

from .errors import RateLimitExceeded
from .util import get_remote_address


class C:
    ENABLED = "RATELIMIT_ENABLED"
    STORAGE_URL = "RATELIMIT_STORAGE_URL"
    STORAGE_OPTIONS = "RATELIMIT_STORAGE_OPTIONS"
    STRATEGY = "RATELIMIT_STRATEGY"
    GLOBAL_LIMITS = "RATELIMIT_GLOBAL"
    SWALLOW_ERRORS = "RATELIMIT_SWALLOW_ERRORS"


class ExtLimit(object):
    """
    simple wrapper to encapsulate limits and their context
    """

    def __init__(self, limit, key_func, scope, per_method, methods, error_message,
                 exempt_when):
        self._limit = limit
        self.key_func = key_func
        self._scope = scope
        self.per_method = per_method
        self.methods = methods and [m.lower() for m in methods] or methods
        self.error_message = error_message
        self.exempt_when = exempt_when

    @property
    def limit(self):
        return self._limit() if callable(self._limit) else self._limit

    @property
    def scope(self):
        return self._scope

    @property
    def is_exempt(self):
        """Check if the limit is exempt."""
        return self.exempt_when and self.exempt_when()


class Limiter(object):
    """
    :param app: :class:`sanic.Sanic` instance to initialize the extension with.
    :param list global_limits: a variable list of strings denoting global
     limits to apply to all routes. :ref:`ratelimit-string` for  more details.
    :param function key_func: a callable that returns the key to rate limit by.
    :param str strategy: the strategy to use. refer to :ref:`ratelimit-strategy`
    :param str storage_uri: the storage location. refer to :ref:`ratelimit-conf`
    :param dict storage_options: kwargs to pass to the storage implementation upon instantiation.
    :param bool swallow_errors: whether to swallow errors when hitting a rate limit.
     An exception will still be logged. default ``False``
    """

    def __init__(self, app=None
                 , key_func=None
                 , global_limits=[]
                 , strategy=None
                 , storage_uri=None
                 , storage_options={}
                 , swallow_errors=False
                 ):
        self.app = app
        self.logger = logging.getLogger("sanic-limiter")

        self.enabled = True
        self._global_limits = []
        self._exempt_routes = set()
        self._request_filters = []
        self._strategy = strategy
        self._storage_uri = storage_uri
        self._storage_options = storage_options
        self._swallow_errors = swallow_errors
        self._key_func = key_func or get_remote_address
        for limit in global_limits:
            self._global_limits.extend(
                [
                    ExtLimit(
                        limit, self._key_func, None, False, None, None, None
                    ) for limit in parse_many(limit)
                    ]
            )
        self._route_limits = {}
        self._dynamic_route_limits = {}
        self._blueprint_dynamic_limits = {}
        self._blueprint_limits = {}
        self._storage = None
        self._limiter = None
        self._storage_dead = False

        class BlackHoleHandler(logging.StreamHandler):
            def emit(*_):
                return

        self.logger.addHandler(BlackHoleHandler())
        if app:
            self.init_app(app)

    def init_app(self, app):
        """
        :param app: :class:`sanic.Sanic` instance to rate limit.
        """
        self.enabled = app.config.setdefault(C.ENABLED, True)
        self._swallow_errors = app.config.setdefault(
            C.SWALLOW_ERRORS, self._swallow_errors
        )
        self._storage_options.update(
            app.config.get(C.STORAGE_OPTIONS, {})
        )
        self._storage = storage_from_string(
            self._storage_uri
            or app.config.setdefault(C.STORAGE_URL, 'memory://'),
            **self._storage_options
        )
        strategy = (
            self._strategy
            or app.config.setdefault(C.STRATEGY, 'fixed-window')
        )
        if strategy not in STRATEGIES:
            raise ConfigurationError("Invalid rate limiting strategy %s" % strategy)
        self._limiter = STRATEGIES[strategy](self._storage)

        conf_limits = app.config.get(C.GLOBAL_LIMITS, None)
        if not self._global_limits and conf_limits:
            self._global_limits = [
                ExtLimit(
                    limit, self._key_func, None, False, None, None, None
                ) for limit in parse_many(conf_limits)
                ]
        app.request_middleware.append(self.__check_request_limit)

    @property
    def limiter(self):
        return self._limiter

    def __check_request_limit(self, request):
        endpoint = request.path or ""
        view_handler = self.app.router.routes_static.get(endpoint, None)
        if view_handler is None:
            return
        view_func = view_handler.handler
        view_bpname = view_func.__dict__.get('__blueprintname__', None)
{'fraction_non_alphanumeric': 0.0802273574357063, 'fraction_numerical': 0.0002795378307864331, 'mean_word_length': 2.237707390648567, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '3596367', 'n_tokens_mistral': 3031, 'n_tokens_neox': 2996, 'n_words': 761}