From 6f9bb314c5136ccab36be85cceed3fc17bfe662b Mon Sep 17 00:00:00 2001
From: Alexander Egorov <mofr@zond.org>
Date: Tue, 27 Oct 2026 14:47:01 +0100
Subject: [PATCH] feat(web)!: throttle in the storage of the application's
 injector, reset through Throttle

Throttling takes its storage from the injector of the application, and an
application with @throttling and no ThrottlingStorage fails on startup. The
adapter gives handle_request the identity of its users: the Django adapter
throttles by the Django user, by the client address otherwise. An API method
with @throttling receives its Throttle, and throttle.reset() forgets what the
scope counted for the identity the request was counted for.

set_throttling_storage, set_throttling_identity_resolver and reset(request,
scope) are gone.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 README.md                             |  9 +++-
 docs/architecture.md                  | 15 +++---
 tests/api/api_with_throttling.py      |  7 ++-
 tests/apps.py                         |  6 ---
 tests/conftest.py                     |  4 ++
 tests/test_application_scope.py       | 32 ++++++++++-
 tests/web/test_throttling.py          | 65 +++++++++++++++++++---
 tests/web/test_throttling_identity.py |  7 ---
 winter/django/view.py                 |  6 +--
 winter/web/__init__.py                |  3 +-
 winter/web/request_handler.py         | 30 +++++++++--
 winter/web/throttling/__init__.py     |  9 ++--
 winter/web/throttling/identity.py     | 13 -----
 winter/web/throttling/storage.py      | 20 +------
 winter/web/throttling/throttling.py   | 77 +++++++++++++++++----------
 15 files changed, 191 insertions(+), 112 deletions(-)

diff --git a/README.md b/README.md
index 603e3439..f450dabe 100644
--- a/README.md
+++ b/README.md
@@ -79,8 +79,13 @@ urlpatterns = winter.django.create_urls(app)
   `get_previous_page_url` and `get_next_page_url` is gone.
 - The request pipeline takes the injector from the application. `winter.application.set_injector` is needed only
   by `winter.sqlalchemy`, `winter.ddd` and `winter.messaging`.
-- `winter.django.create_urls` installs the Django throttling identity on its first call: replace it with
-  `winter.web.set_throttling_identity_resolver` after that.
+- Throttling counts in the storage that the injector of the application binds,
+  e.g. `binder.bind(ThrottlingStorage, to=RedisThrottlingStorage(configuration))`; an application with `@throttling`
+  and no storage fails on startup. The Django adapter throttles by the Django user, by the client address otherwise.
+  `set_throttling_storage` and `set_throttling_identity_resolver` are gone.
+- `winter.web.throttling.reset(request, scope)` is gone: an API method with `@throttling` declares a
+  `throttle: winter.web.Throttle` argument, and `throttle.reset()` forgets what the scope has counted for the sender:
+  for the identity that the request was counted for, even if the method has logged the user in since.
 - OpenAPI: `generate_openapi(app, title=..., version=...)` returns the spec and checks `@raises` annotations
   of the application (`allow_missing_raises_annotation=True` skips the check).
   `winter.openapi.serve_spec(app, spec, url='openapi.json')` and
diff --git a/docs/architecture.md b/docs/architecture.md
index f109e4fd..7b3ac7da 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -84,17 +84,17 @@ at least two of them.
 
 - **Framework behaviour is still process-global.** A `winter.web.WebApplication` holds the code of
   a web application, its injector and its interceptors. The adapters derive its exception handlers
-  from its definitions and handle each of its requests in a child of its injector. The other
-  behavioural registries of `winter.web` and `winter.openapi` (native response types, throttling,
-  OpenAPI inspectors) are module singletons, and
+  from its definitions, handle each of its requests in a child of its injector and throttle in the
+  storage that its injector binds. The other behavioural registries of `winter.web` and
+  `winter.openapi` (native response types, OpenAPI inspectors) are module singletons, and
   `winter.sqlalchemy`, `winter.ddd` and `winter.messaging` take the injector from the service
   locator `winter.application.get_injector`.
 - **Models are initialised once per process.** The Django adapter's `_initialize_process` and
-  `winter.openapi.initialization.initialize` fill those module singletons with the Django entries and
-  the standard inspectors on first use. In the
+  `winter.openapi.initialization.initialize` fill those module singletons with the Django response type
+  and the standard inspectors on first use. In the
   target, a `WebApplication` is given these registries like its interceptors, and they start with the
   standard entries. Building from an application, an adapter adds
-  its own entries (the native response type, the throttling identity) to that application only. The
+  its own entries (the native response type) to that application only. The
   `initialize` functions then disappear.
 - **`winter.wsgi`, `winter.asgi` and `winter.rabbitmq` do not exist yet.** HTTP is served through
   `winter.django` only, and `winter.messaging` has no transport adapter.
@@ -110,7 +110,8 @@ at least two of them.
 - **Isolation test** `tests/test_application_scope.py` builds two applications in one process
   and asserts that the handler of a `@problem` exception of one does not handle that exception in
   the other, that the arguments of an API method come from the injector of the application that
-  serves it, and that an interceptor of one does not run in the other. Each registry that moves into
+  serves it, that an interceptor of one does not run in the other, and that each counts throttled
+  requests in the storage of its own injector. Each registry that moves into
   the application adds its case there.
 - **This document** describes intent and is updated when the intent changes, not when the code
   moves. Deviations are listed above while they exist.
diff --git a/tests/api/api_with_throttling.py b/tests/api/api_with_throttling.py
index 84750e3d..bb87d881 100644
--- a/tests/api/api_with_throttling.py
+++ b/tests/api/api_with_throttling.py
@@ -1,8 +1,7 @@
 
 import winter.web
 from winter.web import route_post, route_get
-from winter.web import Request
-from winter.web.throttling import reset
+from winter.web import Throttle
 
 
 @route_get('with-throttling/')
@@ -33,7 +32,7 @@ def method_for_request_without_user(self) -> int:
 
     @route_get('with-reset/{?is_reset}')
     @winter.web.throttling('5/s', 'reset_scope')
-    def simple_method_with_reset(self, request: Request, is_reset: bool) -> int:
+    def simple_method_with_reset(self, throttle: Throttle, is_reset: bool) -> int:
         if is_reset:
-            reset(request, 'reset_scope')
+            throttle.reset()
         return 1
diff --git a/tests/apps.py b/tests/apps.py
index 9f44a987..b1a64ca0 100644
--- a/tests/apps.py
+++ b/tests/apps.py
@@ -1,8 +1,5 @@
 from django.apps import AppConfig
 
-import winter.web
-from winter.web import InMemoryThrottlingStorage
-
 
 class TestAppConfig(AppConfig):
     name = 'tests'
@@ -10,6 +7,3 @@ class TestAppConfig(AppConfig):
     def ready(self):
         # Importing the URLs creates the application and its URL patterns
         from .urls import urlpatterns   # noqa: F401
-
-        # The Redis storage is covered in tests/redis; the sample application throttles in memory.
-        winter.web.set_throttling_storage(InMemoryThrottlingStorage())
diff --git a/tests/conftest.py b/tests/conftest.py
index df626d11..052e9547 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -8,6 +8,8 @@
 from sqlalchemy.engine import Engine
 
 from winter.application import set_injector
+from winter.web import InMemoryThrottlingStorage
+from winter.web import ThrottlingStorage
 
 
 def pytest_configure():
@@ -44,6 +46,8 @@ def pytest_configure():
 class Configuration(Module):
     def configure(self, binder):
         binder.bind(Engine, to=CallableProvider(make_engine), scope=singleton)
+        # The Redis storage is covered in tests/redis; the sample application throttles in memory
+        binder.bind(ThrottlingStorage, to=InMemoryThrottlingStorage, scope=singleton)
 
 
 def make_engine():
diff --git a/tests/test_application_scope.py b/tests/test_application_scope.py
index 1b5759e0..fcc3fbb2 100644
--- a/tests/test_application_scope.py
+++ b/tests/test_application_scope.py
@@ -2,7 +2,8 @@
 Isolation test for docs/architecture.md, principle 2: framework behaviour is never process-global.
 Two applications in one process must not share behaviour: the handler of a @problem exception of one application
 does not handle that exception in another, the arguments of an API method come from the injector of the
-application that serves it, and an interceptor of one application does not run in another.
+application that serves it, an interceptor of one application does not run in another, and each application counts
+throttled requests in the storage of its own injector.
 """
 from http import HTTPStatus
 
@@ -96,3 +97,32 @@ def test_interceptor_of_one_application_does_not_run_in_another():
     # Assert
     assert response_1.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
     assert response_2.content == b'"ok"'
+
+
+class _ThrottledAPI:
+    @winter.web.route_get('throttled/')
+    @winter.web.throttling('1/m')
+    def get(self) -> str:
+        return 'ok'
+
+
+def _application_with_throttling() -> winter.web.WebApplication:
+    storage = winter.web.InMemoryThrottlingStorage()
+    app = winter.web.WebApplication(
+        modules=[],
+        injector=Injector([lambda binder: binder.bind(winter.web.ThrottlingStorage, to=storage)]),
+    )
+    app.add(_ThrottledAPI)
+    return app
+
+
+def test_throttled_requests_of_one_application_are_not_counted_in_another():
+    [url_1] = winter.django.create_urls(_application_with_throttling())
+    [url_2] = winter.django.create_urls(_application_with_throttling())
+    url_1.callback(RequestFactory().get('/throttled/'))
+
+    # Act
+    response = url_2.callback(RequestFactory().get('/throttled/'))
+
+    # Assert
+    assert response.status_code == HTTPStatus.OK
diff --git a/tests/web/test_throttling.py b/tests/web/test_throttling.py
index d384132e..4efe4e48 100644
--- a/tests/web/test_throttling.py
+++ b/tests/web/test_throttling.py
@@ -3,12 +3,13 @@
 
 import freezegun
 import pytest
-from mock import patch
 
+import winter.django
+import winter.web
 from winter.web import ThrottlingMisconfigurationException
 from winter.web.throttling import InMemoryThrottlingStorage
-from winter.web.throttling import get_throttling_storage
-from winter.web.throttling import storage
+from winter.web.throttling import Throttle
+from winter.web.throttling.throttling import Throttling
 
 expected_error_response = {
     'status': 429,
@@ -71,12 +72,60 @@ def test_get_throttling_with_conditional_reset(api_client):
             assert response.status_code == HTTPStatus.OK, i
 
 
-@patch.object(storage, '_throttling_storage', None)
-def test_get_throttling_storage_without_configuration():
-    with pytest.raises(ThrottlingMisconfigurationException) as exc_info:
-        get_throttling_storage()
+class _ThrottledAPI:
+    @winter.web.route_get('throttled/')
+    @winter.web.throttling('5/s')
+    def get(self):  # pragma: no cover
+        pass
 
-    assert 'Throttling storage must be set' in str(exc_info.value)
+
+def test_throttling_without_storage_in_injector_fails_startup():
+    app = winter.web.WebApplication(modules=[])
+    app.add(_ThrottledAPI)
+
+    # Act
+    with pytest.raises(ThrottlingMisconfigurationException) as exception_info:
+        winter.django.create_urls(app)
+
+    # Assert
+    assert str(exception_info.value) == (
+        '_ThrottledAPI.get: @throttling needs a ThrottlingStorage bound in the injector'
+    )
+
+
+class _NotThrottledAPI:
+    @winter.web.route_get('not-throttled/')
+    def get(self, throttle: Throttle):  # pragma: no cover
+        pass
+
+
+def test_throttle_argument_without_throttling_fails_startup():
+    app = winter.web.WebApplication(modules=[])
+    app.add(_NotThrottledAPI)
+
+    # Act
+    with pytest.raises(TypeError) as exception_info:
+        winter.django.create_urls(app)
+
+    # Assert
+    assert str(exception_info.value) == (
+        '_NotThrottledAPI.get: argument throttle (Throttle) '
+        'is not mapped from the request and has no binding in the injector'
+    )
+
+
+def test_throttle_resets_identity_it_counted_the_request_for():
+    storage = InMemoryThrottlingStorage()
+    identities = iter(['10.0.0.1', '42'])
+    throttling = Throttling(num_requests=1, duration=60, scope='login')
+    throttle = Throttle(throttling, storage, None, lambda request: next(identities))
+    throttle.allow_request()
+
+    # Act
+    throttle.reset()
+
+    # Assert
+    assert storage.is_request_allowed('throttle_login_10.0.0.1', duration=60, num_requests=1)
 
 
 def test_in_memory_storage_sliding_window():
diff --git a/tests/web/test_throttling_identity.py b/tests/web/test_throttling_identity.py
index adefcf6d..529f24f0 100644
--- a/tests/web/test_throttling_identity.py
+++ b/tests/web/test_throttling_identity.py
@@ -1,7 +1,5 @@
 from unittest import mock
 
-from winter.web.throttling import get_throttling_identity
-from winter.web.throttling import identity
 from winter.web.throttling import remote_address_identity
 
 
@@ -20,8 +18,3 @@ def test_remote_address_identity_prefers_x_forwarded_for():
     request = _request(headers={'X-Forwarded-For': '203.0.113.5, 10.0.0.1'})
 
     assert remote_address_identity(request) == '203.0.113.5,10.0.0.1'
-
-
-def test_installed_resolver_decides_the_identity():
-    with mock.patch.object(identity, '_identity_resolver', lambda request: 'api-key'):
-        assert get_throttling_identity(_request()) == 'api-key'
diff --git a/winter/django/view.py b/winter/django/view.py
index 01eede15..7d6c87f2 100644
--- a/winter/django/view.py
+++ b/winter/django/view.py
@@ -12,7 +12,6 @@
 from winter.web import WebApplication
 from winter.web import handle_request
 from winter.web import register_native_response_type
-from winter.web import set_throttling_identity_resolver
 from winter.web.autodiscovery import find_routes
 from winter.web.exception_handlers import find_exception_handlers
 from winter.web.exceptions import ExceptionHandlers
@@ -40,11 +39,10 @@ def create_urls(app: WebApplication) -> List[URLPattern]:
 
 
 def _initialize_process():
-    """Process-global until these registries belong to an application: docs/architecture.md, Current deviations."""
+    """Process-global until native response types belong to an application: docs/architecture.md, Current deviations."""
     global _process_initialized
     if not _process_initialized:
         register_native_response_type(HttpResponseBase)
-        set_throttling_identity_resolver(django_user_identity)
         _process_initialized = True
 
 
@@ -98,7 +96,7 @@ def _create_dispatch_function(app: WebApplication, route: Route, exception_handl
     def dispatch(winter_view, request: django.http.HttpRequest, **path_parameters):
         winter_request = DjangoRequest(request, path_parameters)
         request_values = {django.http.HttpRequest: request}
-        response = handle_request(app, route, winter_request, exception_handlers, request_values)
+        response = handle_request(app, route, winter_request, exception_handlers, request_values, django_user_identity)
         return to_django_response(response)
 
     return dispatch
diff --git a/winter/web/__init__.py b/winter/web/__init__.py
index 3244581e..03d382e0 100644
--- a/winter/web/__init__.py
+++ b/winter/web/__init__.py
@@ -26,10 +26,9 @@
 from .routing import route_post
 from .routing import route_put
 from .throttling import InMemoryThrottlingStorage
+from .throttling import Throttle
 from .throttling import ThrottlingMisconfigurationException
 from .throttling import ThrottlingStorage
-from .throttling import set_throttling_identity_resolver
-from .throttling import set_throttling_storage
 from .throttling import throttling
 from .urls import register_url_regexp
 from .web_application import WebApplication
diff --git a/winter/web/request_handler.py b/winter/web/request_handler.py
index 39c76b01..3c4594de 100644
--- a/winter/web/request_handler.py
+++ b/winter/web/request_handler.py
@@ -21,11 +21,17 @@
 from .request_body_resolver import RequestBodyArgumentResolver
 from .request_header_resolver import RequestHeaderArgumentResolver
 from .request_scope import create_request_injector
+from .request_scope import is_bound
 from .response import Response
 from .response_entity import ResponseEntity
 from .response_header_resolver import ResponseHeaderArgumentResolver
 from .routing import Route
-from .throttling import create_throttle_class
+from .throttling import IdentityResolver
+from .throttling import Throttle
+from .throttling import ThrottlingMisconfigurationException
+from .throttling import ThrottlingStorage
+from .throttling import create_throttle
+from .throttling import get_throttling
 from .web_application import WebApplication
 
 # The arguments mapped from the request; any other argument comes from the injector of the request
@@ -48,11 +54,20 @@ def check_arguments(
 ):
     """
     Fails on startup for an argument of an API method, an interceptor or an exception handler that is not mapped
-    from the request and whose type the injector does not bind. Adapters bind the request types in each request.
+    from the request and whose type the injector does not bind, and for an API method with @throttling when the
+    injector binds no ThrottlingStorage. Adapters bind the request types in each request; an API method with
+    @throttling also receives its Throttle.
     """
     request_types = {Request, *request_types}
     for route in routes:
-        _arguments_resolver.check_arguments(route.method, app.injector, request_types)
+        if get_throttling(route.method) is None:
+            _arguments_resolver.check_arguments(route.method, app.injector, request_types)
+            continue
+        if not is_bound(app.injector, ThrottlingStorage):
+            raise ThrottlingMisconfigurationException(
+                f'{route.method.full_name}: @throttling needs a ThrottlingStorage bound in the injector',
+            )
+        _arguments_resolver.check_arguments(route.method, app.injector, {*request_types, Throttle})
     for interceptor in app.interceptors:
         pre_handle = ComponentMethod.get_or_create(interceptor.pre_handle)
         _arguments_resolver.check_arguments(pre_handle, app.injector, request_types, {'method'})
@@ -66,6 +81,7 @@ def handle_request(
     request: Request,
     exception_handlers: ExceptionHandlers,
     request_values: Mapping[type, object],
+    throttling_identity: IdentityResolver,
 ) -> Response:
     """
     Run the whole handling pipeline for a matched route of the application: throttling, its interceptors, argument
@@ -73,14 +89,18 @@ def handle_request(
     Framework-agnostic: adapters build the Request from their own request object and send the returned Response.
     The request is handled in an injector of its own, a child of the application's one, that binds the Request
     and the request values of the adapter, such as its own request object. Request-scoped values live in it.
+    Throttling counts the request in the storage of the application's injector for the identity that the adapter
+    resolves, since the adapter knows the users of its framework.
     """
     method = route.method
     response_headers: dict[str, str] = {}
+    throttle = create_throttle(method, request, app.injector, throttling_identity)
+    if throttle is not None:
+        request_values = {**request_values, Throttle: throttle}
     injector = create_request_injector(app.injector, request, request_values)
 
-    throttle = create_throttle_class(route)
     try:
-        if throttle and not throttle.allow_request(request):
+        if throttle is not None and not throttle.allow_request():
             raise ThrottleException()
 
         for interceptor_cls in app.interceptors:
diff --git a/winter/web/throttling/__init__.py b/winter/web/throttling/__init__.py
index 26182e01..310b2296 100644
--- a/winter/web/throttling/__init__.py
+++ b/winter/web/throttling/__init__.py
@@ -1,12 +1,9 @@
 from .exceptions import ThrottlingMisconfigurationException
 from .identity import IdentityResolver
-from .identity import get_throttling_identity
 from .identity import remote_address_identity
-from .identity import set_throttling_identity_resolver
 from .storage import InMemoryThrottlingStorage
 from .storage import ThrottlingStorage
-from .storage import get_throttling_storage
-from .storage import set_throttling_storage
-from .throttling import create_throttle_class
-from .throttling import reset
+from .throttling import Throttle
+from .throttling import create_throttle
+from .throttling import get_throttling
 from .throttling import throttling
diff --git a/winter/web/throttling/identity.py b/winter/web/throttling/identity.py
index 09da647d..4ad936a5 100644
--- a/winter/web/throttling/identity.py
+++ b/winter/web/throttling/identity.py
@@ -10,16 +10,3 @@ def remote_address_identity(request: Request) -> str:
     """The client address: X-Forwarded-For when a proxy sets it, the socket address otherwise."""
     forwarded_for = request.headers.get('X-Forwarded-For')
     return ''.join(forwarded_for.split()) if forwarded_for else request.remote_address
-
-
-_identity_resolver: IdentityResolver = remote_address_identity
-
-
-def set_throttling_identity_resolver(resolver: IdentityResolver):
-    """Adapters and applications install the resolver that knows their notion of a user."""
-    global _identity_resolver
-    _identity_resolver = resolver
-
-
-def get_throttling_identity(request: Request) -> str:
-    return _identity_resolver(request)
diff --git a/winter/web/throttling/storage.py b/winter/web/throttling/storage.py
index 6e4a0ae6..549fd263 100644
--- a/winter/web/throttling/storage.py
+++ b/winter/web/throttling/storage.py
@@ -4,13 +4,11 @@
 from collections import defaultdict
 from collections import deque
 
-from .exceptions import ThrottlingMisconfigurationException
-
 
 class ThrottlingStorage(abc.ABC):
     """
-    Sliding-window request counter. Implementations live in adapters (see winter.redis.throttling);
-    InMemoryThrottlingStorage below is for single-process deployments and tests.
+    Sliding-window request counter, bound in the injector of an application that throttles. Implementations live in
+    adapters (see winter.redis.throttling); InMemoryThrottlingStorage below is for single-process deployments and tests.
     """
 
     @abc.abstractmethod
@@ -41,17 +39,3 @@ def is_request_allowed(self, key: str, duration: int, num_requests: int) -> bool
     def delete(self, key: str):
         with self._lock:
             self._timestamps.pop(key, None)
-
-
-_throttling_storage: ThrottlingStorage | None = None
-
-
-def set_throttling_storage(storage: ThrottlingStorage):
-    global _throttling_storage
-    _throttling_storage = storage
-
-
-def get_throttling_storage() -> ThrottlingStorage:
-    if _throttling_storage is None:
-        raise ThrottlingMisconfigurationException('Throttling storage must be set before using the throttling')
-    return _throttling_storage
diff --git a/winter/web/throttling/throttling.py b/winter/web/throttling/throttling.py
index e42dd547..0ecae365 100644
--- a/winter/web/throttling/throttling.py
+++ b/winter/web/throttling/throttling.py
@@ -1,16 +1,14 @@
 import dataclasses
 from typing import Optional
-from typing import TYPE_CHECKING
 from typing import Tuple
 
+from injector import Injector
 
 from winter.web.request import Request
+from winter.components import ComponentMethod
 from winter.components import annotate_method
-from .identity import get_throttling_identity
-from .storage import get_throttling_storage
-
-if TYPE_CHECKING:
-    from winter.web.routing import Route  # noqa: F401
+from .identity import IdentityResolver
+from .storage import ThrottlingStorage
 
 
 @dataclasses.dataclass
@@ -30,26 +28,36 @@ def throttling(rate: Optional[str], scope: Optional[str] = None):
     return annotate_method(ThrottlingAnnotation(rate, scope), single=True)
 
 
-class BaseRateThrottle:
-    def __init__(self, throttling_: Throttling):
-        self._throttling = throttling_
-        self._storage = get_throttling_storage()
-
-    def allow_request(self, request: Request) -> bool:
-        ident = get_throttling_identity(request)
-        key = _get_cache_key(self._throttling.scope, ident)
+class Throttle:
+    """
+    The throttling of one request in the scope of its API method. An API method with @throttling receives it as an
+    argument to reset what is counted for whoever sends the request.
+    """
 
-        return self._storage.is_request_allowed(key, self._throttling.duration, self._throttling.num_requests)
+    def __init__(
+        self,
+        throttling_: Throttling,
+        storage: ThrottlingStorage,
+        request: Request,
+        identity: IdentityResolver,
+    ):
+        self._throttling = throttling_
+        self._storage = storage
+        self._request = request
+        self._identity = identity
+        self._key: Optional[str] = None
 
+    def allow_request(self) -> bool:
+        """Count the request unless the scope has reached its rate for whoever sends it."""
+        self._key = _get_cache_key(self._throttling.scope, self._identity(self._request))
+        return self._storage.is_request_allowed(self._key, self._throttling.duration, self._throttling.num_requests)
 
-def reset(request: Request, scope: str):
-    """
-        This function allows to reset the accumulated throttling state
-        for a specific user and scope
-    """
-    ident = get_throttling_identity(request)
-    key = _get_cache_key(scope, ident)
-    get_throttling_storage().delete(key)
+    def reset(self):
+        """
+        Forget the requests counted in the scope for whoever sends the request, this one included. The identity is
+        the one the request was counted for, even if the API method has logged the user in since.
+        """
+        self._storage.delete(self._key)
 
 
 CACHE_KEY_FORMAT = 'throttle_{scope}_{ident}'
@@ -70,14 +78,25 @@ def _parse_rate(rate: str) -> Tuple[int, int]:
     return num_requests, duration
 
 
-def create_throttle_class(route: 'Route') -> Optional[BaseRateThrottle]:
-    throttling_annotation = route.method.annotations.get_one_or_none(ThrottlingAnnotation)
+def get_throttling(method: ComponentMethod) -> Optional[Throttling]:
+    throttling_annotation = method.annotations.get_one_or_none(ThrottlingAnnotation)
 
     if getattr(throttling_annotation, 'rate', None) is None:
         return None
 
     num_requests, duration = _parse_rate(throttling_annotation.rate)
-    throttling_scope = throttling_annotation.scope or route.method.full_name
-    throttling_ = Throttling(num_requests, duration, throttling_scope)
-
-    return BaseRateThrottle(throttling_)
+    throttling_scope = throttling_annotation.scope or method.full_name
+    return Throttling(num_requests, duration, throttling_scope)
+
+
+def create_throttle(
+    method: ComponentMethod,
+    request: Request,
+    app_injector: Injector,
+    identity: IdentityResolver,
+) -> Optional[Throttle]:
+    """The throttle of a request to the method, counting in the storage that the injector of the application binds."""
+    throttling_ = get_throttling(method)
+    if throttling_ is None:
+        return None
+    return Throttle(throttling_, app_injector.get(ThrottlingStorage), request, identity)