A rate limiting module for Ellar
Project description
Ellar - Python ASGI web framework for building fast, efficient, and scalable RESTful APIs and server-side applications.
Full Documentation: Here
Introduction
A rate limit module for Ellar
Installation
$(venv) pip install ellar-throttler
Usage
ThrottlerModule
The ThrottleModule
is the main entry point for this package, and can be used in a synchronous or asynchronous manner.
All the needs to be passed is the ttl
, the time to live in seconds for the request tracker, and the limit
,
or how many times an endpoint can be hit before returning a 429 status code.
from ellar.common import Module
from ellar_throttler import ThrottlerModule
@Module(modules=[
ThrottlerModule.setup(ttl=60, limit=10)
])
class ApplicationModule:
pass
The above would mean that 10 requests from the same IP can be made to a single endpoint in 1 minute.
from ellar.common import Module
from ellar_throttler import ThrottlerModule, ThrottlerGuard
from ellar.core import Config, ModuleSetup, DynamicModule
def throttler_module_factory(module: ThrottlerModule, config: Config) -> DynamicModule:
return module.setup(ttl=config['THROTTLE_TTL'], limit=config['THROTTLE_LIMIT'])
@Module(modules=[
ModuleSetup(ThrottlerModule, inject=[Config], factory=throttler_module_factory)
])
class ApplicationModule:
pass
# server.py
application = AppFactory.create_from_app_module(
ApplicationModule,
config_module=os.environ.get(
ELLAR_CONFIG_MODULE, "dialerai.config:DevelopmentConfig"
),
global_guards=[ThrottlerGuard]
)
The above is also a valid configuration for ThrottleModule
registration if you want to work with config.
NOTE: If you add the ThrottlerGuard
to your application global_guards
, then all the incoming requests will be throttled by default.
This can also be omitted in favor of @UseGuards(ThrottlerGuard)
.
The global guard check can be skipped using the @skip_throttle()
decorator mentioned later.
Example with @UseGuards(ThrottlerGuard)
# project_name/controller.py
from ellar.common import Controller, UseGuards
from ellar_throttler import throttle, ThrottlerGuard
@Controller()
class AppController:
@UseGuards(ThrottlerGuard)
@throttle(limit=5, ttl=30)
def normal(self):
pass
Decorators
@throttle()
@throttle(*, limit: int = 20, ttl: int = 60)
This decorator will set THROTTLER_LIMIT
and THROTTLER_TTL
metadata on the route, for retrieval from the Reflector class.
It can be applied to controllers and routes.
@skip_throttle()
@skip_throttle(skip: bool = True)
This decorator can be used to skip a route or a class or to negate the skipping of a route in a class that is skipped.
# project_name/controller.py
from ellar.common import Controller, UseGuards
from ellar_throttler import ThrottlerGuard, skip_throttle
@skip_throttle()
@UseGuards(ThrottlerGuard)
@Controller()
class AppController:
def do_skip(self):
pass
@skip_throttle(skip=False)
def dont_skip(self):
pass
In the above controller, dont_skip
would be counted against and
rate-limited while do_skip
would not be limited in any way.
ThrottlerStorage
Interface to define the methods to handle the details when it comes to keeping track of the requests.
Currently, the key is seen as an MD5
hash of the IP the class name
and the function name
,
to ensure that no unsafe characters are used.
The interface looks like this:
import typing as t
from abc import ABC, abstractmethod
class IThrottlerStorage(ABC):
@property
@abstractmethod
def storage(self) -> t.Dict[str, ThrottlerStorageOption]:
"""
The internal storage with all the request records.
The key is a hashed key based on the current context and IP.
:return:
"""
@abstractmethod
async def increment(self, key: str, ttl: int) -> ThrottlerStorageRecord:
"""
Increment the amount of requests for a given record. The record will
automatically be removed from the storage once its TTL has been reached.
:param key:
:param ttl:
:return:
"""
So long as the Storage service implements this interface, it should be usable by the ThrottlerGuard
.
Proxies
If you are working with multiple proxies, you can override the get_tracker()
method to pull the value from the header or install
ProxyHeadersMiddleware
# throttler_behind_proxy.guard.py
from ellar_throttler import ThrottlerGuard
from ellar.di import injectable
from ellar.core.connection import HTTPConnection
@injectable()
class ThrottlerBehindProxyGuard(ThrottlerGuard):
def get_tracker(self, connection: HTTPConnection) -> str:
return connection.client.host # individualize IP extraction to meet your own needs
# project_name/controller.py
from .throttler_behind_proxy import ThrottlerBehindProxyGuard
@Controller('')
@UseGuards(ThrottlerBehindProxyGuard)
class AppController:
pass
Working with WebSockets
To work with Websockets you can extend the ThrottlerGuard
and override the handle_request
method with the code below:
from ellar_throttler import ThrottlerGuard
from ellar.di import injectable
from ellar.common import IExecutionContext
from ellar_throttler import ThrottledException
@injectable()
class WsThrottleGuard(ThrottlerGuard):
async def handle_request(self, context: IExecutionContext, limit: int, ttl: int) -> bool:
websocket_client = context.switch_to_websocket().get_client()
host = websocket_client.client.host
key = self.generate_key(context, host)
result = await self.storage_service.increment(key, ttl)
# Throw an error when the user reached their limit.
if result.total_hits > limit:
raise ThrottledException(wait=result.time_to_expire)
return True
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distribution
Hashes for ellar_throttler-0.0.9-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | af4826249b03575db4224cb790ec581a84b410b897ea4ee8c966cbafa4feee85 |
|
MD5 | cec6787e246ceee3c18f4accc87b28ab |
|
BLAKE2b-256 | 85cf94cfe1431e0d2391d06f6badd8da12778b7fc7a1cab7716ec7a33666a169 |