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
814 Jan 08, 2023
74 Dec 30, 2022
1k Jan 04, 2023
19 Jul 23, 2022
6.2k Jan 01, 2023
1 Oct 22, 2021
6k Jan 03, 2023
17 Oct 18, 2022
4 Mar 23, 2022
927 Jan 08, 2023
5.7k Jan 08, 2023
33.8k Jan 09, 2023
85 Dec 29, 2022
9 Jul 22, 2022
6.7k Dec 27, 2022
1 Jan 04, 2022
1 Nov 08, 2021
526 Jan 02, 2023
13.2k Jan 05, 2023