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
4 Aug 25, 2022
491 Dec 08, 2022
2 Dec 03, 2021
9 Feb 04, 2022
4k Jan 08, 2023
91 Nov 10, 2022
525 Jan 01, 2023
38 Jul 25, 2022
119 Dec 27, 2022
85 Dec 29, 2022
3.8k Jan 02, 2023
16 Dec 15, 2022
927 Jan 08, 2023
1k Jan 04, 2023
814 Jan 08, 2023
32 Dec 24, 2022
94 Nov 20, 2022
46 Nov 08, 2022
32 Mar 05, 2022
26 Sep 11, 2022