aiohttp-ratelimiter
aiohttp-ratelimiter is a rate limiter for the aiohttp.web framework. This is a new library and we are always looking for people to contribute. If you see something wrong with the code or want to add a feature, please create a pull request on our github.
Install from git
python -m pip install git+https://github.com/JGLTechnologies/aiohttp-ratelimiter
Install from pypi
python -m pip install aiohttp-ratelimiter
Example
from aiohttp import web
from aiohttplimiter import default_keyfunc, Limiter
app = web.Application()
routes = web.RouteTableDef()
limiter = Limiter(keyfunc=default_keyfunc)
@routes.get("/")
# This endpoint can only be requested 1 time per second per IP address
@limiter.limit("1/1")
async def home(request):
return web.Response(text="test")
app.add_routes(routes)
web.run_app(app)
You can exempt an IP from rate limiting using the exempt_ips kwarg.
from aiohttplimiter import Limiter, default_keyfunc
from aiohttp import web
app = web.Application()
routes = web.RouteTableDef()
# 192.168.1.245 is exempt from rate limiting.
# Keep in mind that exempt_ips takes a set not a list.
limiter = Limiter(keyfunc=default_keyfunc, exempt_ips={"192.168.1.245"})
@routes.get("/")
@limiter.limit("1/1")
async def test(request):
return web.Response(text="test")
app.add_routes(routes)
web.run_app(app)
You can create your own error handler by using the error_handler kwarg.
from aiohttplimiter import Allow, RateLimitExceeded, Limiter, default_keyfunc
from aiohttp import web
def handler(request: web.Request, exc: RateLimitExceeded):
# If for some reason you want to allow the request, return aiohttplimitertest.Allow().
if some_condition:
return Allow()
return web.Response(text="Too many requests", status=429)
limiter = Limiter(keyfunc=default_keyfunc, error_handler=handler)
If multiple paths use one handler like this:
@routes.get("/")
@routes.get("/home")
@limiter.limit("5/1")
def home(request):
return web.Response(text="Hello")
Then they will have separate rate limits. To prevent this use the path_id kwarg.
@routes.get("/")
@routes.get("/home")
@limiter.limit("5/1", path_id="home")
def home(request):
return web.Response(text="Hello")
If you want to use Redis instead, use the RedisLimiter class.
from aiohttplimiter import RedisLimiter, default_keyfunc
limiter = RedisLimiter(keyfunc=default_keyfunc, uri="redis://:[email protected]:port")
RedisLimiter is still being tested and might not be stable for production.
3.7k Dec 30, 2022
1.9k Jan 4, 2023
2k Dec 31, 2022
796 Dec 26, 2022
687 Nov 27, 2022
264 Dec 31, 2022
83 Sep 23, 2022
1 Nov 16, 2021
162 Dec 26, 2022
7 Nov 02, 2022
32 Dec 14, 2022
1.6k Jan 09, 2023
6.7k Dec 27, 2022
3.8k Jan 02, 2023
3 Apr 12, 2022
8 Jul 09, 2022
16.7k Dec 28, 2022
1 Nov 22, 2021
3 Jun 13, 2022
16 Dec 15, 2022
27 Sep 08, 2022
2 Dec 03, 2021
13.2k Jan 05, 2023
40 Dec 02, 2022
67.9k Dec 29, 2022
78 Apr 15, 2022