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
37 Aug 07, 2022
4k Jan 08, 2023
122 Dec 22, 2022
7 Nov 02, 2022
8 Jul 09, 2022
9 Jul 22, 2022
5.7k Jan 08, 2023
1 Nov 08, 2021
32 Dec 24, 2022
6.7k Dec 27, 2022
2 Jan 04, 2022
7 Dec 14, 2022
552 Dec 30, 2022
61 Apr 06, 2022
61.5k Jan 06, 2023
8 Dec 07, 2021
38 Jul 25, 2022
2 Dec 02, 2021