FastAPI-Limiter is a rate limiting tool for fastapi routes with lua script.
Just install from pypi
> pip install fastapi-limiterFastAPI-Limiter is simple to use, which just provide a dependency RateLimiter, the following example allow 2 times request per 5 seconds in route /.
importredis.asyncioasredisimportuvicornfromcontextlibimportasynccontextmanagerfromfastapiimportDepends, FastAPIfromfastapi_limiterimportFastAPILimiterfromfastapi_limiter.dependsimportRateLimiter@asynccontextmanagerasyncdeflifespan(_: FastAPI): redis_connection=redis.from_url("redis://localhost:6379", encoding="utf8") awaitFastAPILimiter.init(redis_connection) yieldawaitFastAPILimiter.close() app=FastAPI(lifespan=lifespan) @app.get("/", dependencies=[Depends(RateLimiter(times=2, seconds=5))])asyncdefindex(): return{"msg": "Hello World"} if__name__=="__main__": uvicorn.run("main:app", debug=True, reload=True)There are some config in FastAPILimiter.init.
The redis instance of aioredis.
Prefix of redis key.
Identifier of route limit, default is ip, you can override it such as userid and so on.
asyncdefdefault_identifier(request: Request): forwarded=request.headers.get("X-Forwarded-For") ifforwarded: returnforwarded.split(",")[0] returnrequest.client.host+":"+request.scope["path"]Callback when access is forbidden, default is raise HTTPException with 429 status code.
asyncdefdefault_callback(request: Request, response: Response, pexpire: int): """ default callback when too many requests :param request: :param pexpire: The remaining milliseconds :param response: :return: """expire=ceil(pexpire/1000) raiseHTTPException( HTTP_429_TOO_MANY_REQUESTS, "Too Many Requests", headers={"Retry-After": str(expire)} )You can use multiple limiters in one route.
@app.get("/multiple",dependencies=[Depends(RateLimiter(times=1, seconds=5)),Depends(RateLimiter(times=2, seconds=15)), ],)asyncdefmultiple(): return{"msg": "Hello World"}Not that you should note the dependencies orders, keep lower of result of seconds/times at the first.
While the above examples work with rest requests, FastAPI also allows easy usage of websockets, which require a slightly different approach.
Because websockets are likely to be long lived, you may want to rate limit in response to data sent over the socket.
You can do this by rate limiting within the body of the websocket handler:
@app.websocket("/ws")asyncdefwebsocket_endpoint(websocket: WebSocket): awaitwebsocket.accept() ratelimit=WebSocketRateLimiter(times=1, seconds=5) whileTrue: try: data=awaitwebsocket.receive_text() awaitratelimit(websocket, context_key=data) # NB: context_key is optionalawaitwebsocket.send_text(f"Hello, world") exceptWebSocketRateLimitException: # Thrown when rate limit exceeded.awaitwebsocket.send_text(f"Hello again")The lua script used.
localkey=KEYS[1] locallimit=tonumber(ARGV[1]) localexpire_time=ARGV[2] localcurrent=tonumber(redis.call('get', key) or"0") ifcurrent>0thenifcurrent+1>limitthenreturnredis.call("PTTL", key) elseredis.call("INCR", key) return0endelseredis.call("SET", key, 1, "px", expire_time) return0endThis project is licensed under the Apache-2.0 License.