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
198 Jan 07, 2023
525 Jan 01, 2023
11.9k Jan 01, 2023
7.7k Jan 01, 2023
1 Dec 07, 2021
4 Dec 21, 2022
85 Dec 29, 2022
1 Dec 26, 2021
4 Aug 25, 2022
4k Jan 08, 2023
6.7k Dec 27, 2022
37 Aug 07, 2022
9k Jan 01, 2023
53k Jan 02, 2023
293 Dec 31, 2022
5.8k Dec 30, 2022
91 Nov 10, 2022