Source code for

import time
import urllib.parse

from deprecated.sphinx import versionadded

from import Storage
from limits.typing import EmcacheClientP, Optional, Tuple, Type, Union

[docs] @versionadded(version="2.1") class MemcachedStorage(Storage): """ Rate limit storage with memcached as backend. Depends on :pypi:`emcache` """ STORAGE_SCHEME = ["async+memcached"] """The storage scheme for memcached to be used in an async context""" DEPENDENCIES = ["emcache"] def __init__( self, uri: str, wrap_exceptions: bool = False, **options: Union[float, str, bool], ) -> None: """ :param uri: memcached location of the form ``async+memcached://host:port,host:port`` :param wrap_exceptions: Whether to wrap storage exceptions in :exc:`limits.errors.StorageError` before raising it. :param options: all remaining keyword arguments are passed directly to the constructor of :class:`emcache.Client` :raise ConfigurationError: when :pypi:`emcache` is not available """ parsed = urllib.parse.urlparse(uri) self.hosts = [] for host, port in ( loc.split(":") for loc in parsed.netloc.strip().split(",") if loc.strip() ): self.hosts.append((host, int(port))) self._options = options self._storage = None super().__init__(uri, wrap_exceptions=wrap_exceptions, **options) self.dependency = self.dependencies["emcache"].module @property def base_exceptions( self, ) -> Union[Type[Exception], Tuple[Type[Exception], ...]]: # pragma: no cover return ( self.dependency.ClusterNoAvailableNodes, self.dependency.CommandError, ) async def get_storage(self) -> EmcacheClientP: if not self._storage: self._storage = await self.dependency.create_client( [self.dependency.MemcachedHostAddress(h, p) for h, p in self.hosts], **self._options, ) assert self._storage return self._storage
[docs] async def get(self, key: str) -> int: """ :param key: the key to get the counter value for """ item = await (await self.get_storage()).get(key.encode("utf-8")) return item and int(item.value) or 0
[docs] async def clear(self, key: str) -> None: """ :param key: the key to clear rate limits for """ await (await self.get_storage()).delete(key.encode("utf-8"))
[docs] async def incr( self, key: str, expiry: int, elastic_expiry: bool = False, amount: int = 1 ) -> int: """ increments the counter for a given rate limit key :param key: the key to increment :param expiry: amount in seconds for the key to expire in :param elastic_expiry: whether to keep extending the rate limit window every hit. :param amount: the number to increment by """ storage = await self.get_storage() limit_key = key.encode("utf-8") expire_key = f"{key}/expires".encode() added = True try: await storage.add(limit_key, f"{amount}".encode(), exptime=expiry) except self.dependency.NotStoredStorageCommandError: added = False storage = await self.get_storage() if not added: value = await storage.increment(limit_key, amount) or amount if elastic_expiry: await storage.touch(limit_key, exptime=expiry) await storage.set( expire_key, str(expiry + time.time()).encode("utf-8"), exptime=expiry, noreply=False, ) return value else: await storage.set( expire_key, str(expiry + time.time()).encode("utf-8"), exptime=expiry, noreply=False, ) return amount
[docs] async def get_expiry(self, key: str) -> int: """ :param key: the key to get the expiry for """ storage = await self.get_storage() item = await storage.get(f"{key}/expires".encode()) return int(item and float(item.value) or time.time())
[docs] async def check(self) -> bool: """ Check if storage is healthy by calling the ``get`` command on the key ``limiter-check`` """ try: storage = await self.get_storage() await storage.get(b"limiter-check") return True except: # noqa return False
[docs] async def reset(self) -> Optional[int]: raise NotImplementedError