[docs]classWindowStats(NamedTuple):""" Tuple to describe a rate limited window """#: Time as seconds since the Epoch when this window will be resetreset_time:int#: Quantity remaining in this windowremaining:int
@dataclasses.dataclassclassDependency:name:strversion_required:Optional[Version]version_found:Optional[Version]module:ModuleTypeifTYPE_CHECKING:_UserDict=UserDict[str,Dependency]else:_UserDict=UserDictclassDependencyDict(_UserDict):Missing=Dependency("Missing",None,None,ModuleType("Missing"))def__getitem__(self,key:str)->Dependency:dependency=super().__getitem__(key)ifdependency==DependencyDict.Missing:raiseConfigurationError(f"{key} prerequisite not available")elifdependency.version_requiredand(notdependency.version_foundordependency.version_found<dependency.version_required):raiseConfigurationError(f"The minimum version of {dependency.version_required}"f" of {dependency.name} could not be found")returndependencyclassLazyDependency:""" Simple utility that provides an :attr:`dependency` to the child class to fetch any dependencies without having to import them explicitly. """DEPENDENCIES:Union[Dict[str,Optional[Version]],List[str]]=[]""" The python modules this class has a dependency on. Used to lazily populate the :attr:`dependencies` """def__init__(self)->None:self._dependencies:DependencyDict=DependencyDict()@propertydefdependencies(self)->DependencyDict:""" Cached mapping of the modules this storage depends on. This is done so that the module is only imported lazily when the storage is instantiated. :meta private: """ifnotgetattr(self,"_dependencies",None):dependencies=DependencyDict()mapping:Dict[str,Optional[Version]]ifisinstance(self.DEPENDENCIES,list):mapping={dependency:Nonefordependencyinself.DEPENDENCIES}else:mapping=self.DEPENDENCIESforname,minimum_versioninmapping.items():dependency,version=get_dependency(name)ifnotdependency:dependencies[name]=DependencyDict.Missingelse:dependencies[name]=Dependency(name,minimum_version,version,dependency)self._dependencies=dependenciesreturnself._dependenciesdefget_dependency(module_path:str)->Tuple[Optional[ModuleType],Optional[Version]]:""" safe function to import a module at runtime """try:ifmodule_pathnotinsys.modules:__import__(module_path)root=module_path.split(".")[0]version=getattr(sys.modules[root],"__version__","0.0.0")returnsys.modules[module_path],Version(version)exceptImportError:# pragma: no coverreturnNone,Nonedefget_package_data(path:str)->bytes:returncast(bytes,importlib_resources.files("limits").joinpath(path).read_bytes())
[docs]defparse_many(limit_string:str)->List[RateLimitItem]:""" parses rate limits in string notation containing multiple rate limits (e.g. ``1/second; 5/minute``) :param limit_string: rate limit string using :ref:`ratelimit-string` :raise ValueError: if the string notation is invalid. """ifnot(isinstance(limit_string,str)andEXPR.match(limit_string)):raiseValueError("couldn't parse rate limit string '%s'"%limit_string)limits=[]forlimitinSEPARATORS.split(limit_string):match=SINGLE_EXPR.match(limit)ifmatch:amount,_,multiples,granularity_string=match.groups()granularity=granularity_from_string(granularity_string)limits.append(granularity(int(amount),multiplesandint(multiples)orNone))returnlimits
[docs]defparse(limit_string:str)->RateLimitItem:""" parses a single rate limit in string notation (e.g. ``1/second`` or ``1 per second``) :param limit_string: rate limit string using :ref:`ratelimit-string` :raise ValueError: if the string notation is invalid. """returnlist(parse_many(limit_string))[0]
defgranularity_from_string(granularity_string:str)->Type[RateLimitItem]:""" :param granularity_string: :raise ValueError: """forgranularityinGRANULARITIES.values():ifgranularity.check_granularity_string(granularity_string):returngranularityraiseValueError("no granularity matched for %s"%granularity_string)