Coverage for /usr/local/lib/python3.10/site-packages/opal_server-0.0.0-py3.10.egg/opal_server/loadlimiting.py: 67%

1from fastapi import APIRouter, Request 

2from opal_common.logger import logger 

3from slowapi import Limiter 

4 

5 

6def init_loadlimit_router(loadlimit_notation: str = None): 

7 """Initializes a route where a client (or any other network peer) can 

8 inquire what opal clients are currently connected to the server and on what 

9 topics are they registered. 

10 

11 If the OPAL server does not have statistics enabled, the route will 

12 return 501 Not Implemented 

13 """ 

14 router = APIRouter() 

15 

16 # We want to globally limit the endpoint, not per client 

17 limiter = Limiter(key_func=lambda: "global") 

18 

19 if loadlimit_notation: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true

20 logger.info(f"rate limiting is on, configured limit: {loadlimit_notation}") 

21 

22 @router.get("/loadlimit") 

23 @limiter.limit(loadlimit_notation) 

24 async def loadlimit(request: Request): 

25 return 

26 

27 else: 

28 

29 @router.get("/loadlimit") 

30 async def loadlimit(request: Request): 

31 return 

32 

33 return router