[docs]classRequestLimit:""" Provides details of a rate limit within the context of a request """#: The instance of the rate limitlimit:RateLimitItem#: The full key for the request against which the rate limit is testedkey:str#: Whether the limit was breached within the context of this requestbreached:bool#: Whether the limit is a shared limitshared:booldef__init__(self,extension:Limiter,limit:RateLimitItem,request_args:list[str],breached:bool,shared:bool,)->None:self.extension:weakref.ProxyType[Limiter]=weakref.proxy(extension)self.limit=limitself.request_args=request_argsself.key=limit.key_for(*request_args)self.breached=breachedself.shared=sharedself._window:WindowStats|None=None@propertydeflimiter(self)->RateLimiter:returntyping.cast(RateLimiter,self.extension.limiter)@propertydefwindow(self)->WindowStats:ifnotself._window:self._window=self.limiter.get_window_stats(self.limit,*self.request_args)returnself._window@propertydefreset_at(self)->int:"""Timestamp at which the rate limit will be reset"""returnint(self.window[0]+1)@propertydefremaining(self)->int:"""Quantity remaining for this rate limit"""returnself.window[1]
@dataclasses.dataclass(eq=True,unsafe_hash=True)classLimit:""" simple wrapper to encapsulate limits and their context """limit:RateLimitItemkey_func:Callable[[],str]_scope:str|Callable[[str],str]|Noneper_method:bool=Falsemethods:tuple[str,...]|None=Noneerror_message:str|None=Noneexempt_when:Callable[[],bool]|None=Noneoverride_defaults:bool|None=Falsededuct_when:Callable[[Response],bool]|None=Noneon_breach:Callable[[RequestLimit],Response|None]|None=None_cost:Callable[[],int]|int=1shared:bool=Falsedef__post_init__(self)->None:ifself.methods:self.methods=tuple([k.lower()forkinself.methods])@propertydefis_exempt(self)->bool:"""Check if the limit is exempt."""ifself.exempt_when:returnself.exempt_when()returnFalse@propertydefscope(self)->str|None:return(self._scope(request.endpointor"")ifcallable(self._scope)elseself._scope)@propertydefcost(self)->int:ifisinstance(self._cost,int):returnself._costreturnself._cost()@propertydefmethod_exempt(self)->bool:"""Check if the limit is not applicable for this method"""returnself.methodsisnotNoneandrequest.method.lower()notinself.methodsdefscope_for(self,endpoint:str,method:str|None)->str:""" Derive final bucket (scope) for this limit given the endpoint and request method. If the limit is shared between multiple routes, the scope does not include the endpoint. """limit_scope=self.scopeiflimit_scope:ifself.shared:scope=limit_scopeelse:scope=f"{endpoint}:{limit_scope}"else:scope=endpointifself.per_method:assertmethodscope+=f":{method.upper()}"returnscope@dataclasses.dataclass(eq=True,unsafe_hash=True)classLimitGroup:""" represents a group of related limits either from a string or a callable that returns one """limit_provider:Callable[[],str]|strkey_function:Callable[[],str]scope:str|Callable[[str],str]|None=Nonemethods:tuple[str,...]|None=Noneerror_message:str|None=Noneexempt_when:Callable[[],bool]|None=Noneoverride_defaults:bool|None=Falsededuct_when:Callable[[Response],bool]|None=Noneon_breach:Callable[[RequestLimit],Response|None]|None=Noneper_method:bool=Falsecost:Callable[[],int]|int|None=Noneshared:bool=Falsedef__iter__(self)->Iterator[Limit]:limit_str=(self.limit_provider()ifcallable(self.limit_provider)elseself.limit_provider)limit_items=parse_many(limit_str)iflimit_strelse[]forlimitinlimit_items:yieldLimit(limit,self.key_function,self.scope,self.per_method,self.methods,self.error_message,self.exempt_when,self.override_defaults,self.deduct_when,self.on_breach,self.costor1,self.shared,)