Added async logic

Author: PhilippTh

AUTHORS

@@ -32,6 +32,7 @@ Merijn Bertels <merijn.bertels@gmail.com>
 Omer Katz <omer.drow@gmail.com>
 Petr Knap <dev@petrknap.cz>
 Philip Neustrom <philipn@gmail.com>
+Philipp Thumfart <philipp@thumfart.eu>
 Pierre-Olivier Marec <pomarec@free.fr>
 Roman Krejcik <farin@farin.cz>
 Silvan Spross <silvan.spross@gmail.com>

constance/backends/__init__.py

@@ -9,13 +9,31 @@ def get(self, key):
         """
         raise NotImplementedError
 
+    async def aget(self, key):
+        """
+        Get the key from the backend store and return the value.
+        Return None if not found.
+        """
+        raise NotImplementedError
+
     def mget(self, keys):
         """
         Get the keys from the backend store and return a list of the values.
         Return an empty list if not found.
         """
         raise NotImplementedError
 
+    async def amget(self, keys):
+        """
+        Get the keys from the backend store and return a list of the values.
+        Return an empty list if not found.
+        """
+        raise NotImplementedError
+
     def set(self, key, value):
         """Add the value to the backend store given the key."""
         raise NotImplementedError
+
+    async def aset(self, key, value):
+        """Add the value to the backend store given the key."""
+        raise NotImplementedError

constance/backends/database.py

@@ -85,6 +85,46 @@ def get(self, key):
                     self._cache.add(key, value)
         return value
 
+    async def aget(self, key):
+        from asgiref.sync import sync_to_async
+
+        prefixed_key = self.add_prefix(key)
+        value = None
+        if self._cache:
+            value = await self._cache.aget(prefixed_key)
+            if value is None:
+                await sync_to_async(self.autofill)()
+                value = await self._cache.aget(prefixed_key)
+        if value is None:
+            match = await self._model._default_manager.filter(key=prefixed_key).only("value").afirst()
+            if match:
+                value = loads(match.value)
+                if self._cache:
+                    await self._cache.aadd(prefixed_key, value)
+        return value
+
+    async def amget(self, keys):
+        if not keys:
+            return {}
+
+        prefixed_keys_map = {self.add_prefix(key): key for key in keys}
+        results = {}
+
+        if self._cache:
+            cache_results = await self._cache.aget_many(prefixed_keys_map.keys())
+            for prefixed_key, value in cache_results.items():
+                results[prefixed_keys_map[prefixed_key]] = value
+
+        missing_prefixed_keys = [k for k in prefixed_keys_map if prefixed_keys_map[k] not in results]
+        if missing_prefixed_keys:
+            try:
+                async for const in self._model._default_manager.filter(key__in=missing_prefixed_keys):
+                    results[prefixed_keys_map[const.key]] = loads(const.value)
+            except (OperationalError, ProgrammingError):
+                pass
+
+        return results
+
     def set(self, key, value):
         key = self.add_prefix(key)
         created = False
@@ -119,6 +159,13 @@ def set(self, key, value):
 
         signals.config_updated.send(sender=config, key=key, old_value=old_value, new_value=value)
 
+    async def aset(self, key, value):
+        from asgiref.sync import sync_to_async
+
+        # We use sync_to_async because Django's transaction.atomic() and database connections are thread-local.
+        # This ensures the operation runs in the correct database thread until native async transactions are supported.
+        return await sync_to_async(self.set)(key, value)
+
     def clear(self, sender, instance, created, **kwargs):
         if self._cache and not created:
             keys = [self.add_prefix(k) for k in settings.CONFIG]

constance/backends/memory.py

@@ -19,6 +19,10 @@ def get(self, key):
         with self._lock:
             return self._storage.get(key)
 
+    async def aget(self, key):
+        # Memory operations are fast enough that we don't need true async here
+        return self.get(key)
+
     def mget(self, keys):
         if not keys:
             return None
@@ -30,8 +34,18 @@ def mget(self, keys):
                     result.append((key, value))
         return result
 
+    async def amget(self, keys):
+        if not keys:
+            return {}
+        with self._lock:
+            return {key: self._storage[key] for key in keys if key in self._storage}
+
     def set(self, key, value):
         with self._lock:
             old_value = self._storage.get(key)
             self._storage[key] = value
             signals.config_updated.send(sender=config, key=key, old_value=old_value, new_value=value)
+
+    async def aset(self, key, value):
+        # Memory operations are fast enough that we don't need true async here
+        self.set(key, value)

constance/backends/redisd.py

@@ -1,3 +1,4 @@
+import asyncio
 from threading import RLock
 from time import monotonic
 
@@ -19,16 +20,28 @@ def __init__(self):
         connection_cls = settings.REDIS_CONNECTION_CLASS
         if connection_cls is not None:
             self._rd = utils.import_module_attr(connection_cls)()
+            self._ard = self._rd
         else:
             try:
                 import redis
             except ImportError:
                 raise ImproperlyConfigured("The Redis backend requires redis-py to be installed.") from None
+
             if isinstance(settings.REDIS_CONNECTION, str):
                 self._rd = redis.from_url(settings.REDIS_CONNECTION)
             else:
                 self._rd = redis.Redis(**settings.REDIS_CONNECTION)
 
+            try:
+                import redis.asyncio as aredis
+
+                if isinstance(settings.REDIS_CONNECTION, str):
+                    self._ard = aredis.from_url(settings.REDIS_CONNECTION)
+                else:
+                    self._ard = aredis.Redis(**settings.REDIS_CONNECTION)
+            except ImportError:
+                self._ard = self._rd
+
     def add_prefix(self, key):
         return f"{self._prefix}{key}"
 
@@ -38,6 +51,15 @@ def get(self, key):
             return loads(value)
         return None
 
+    async def aget(self, key):
+        if hasattr(self._ard, "aget"):
+            value = await self._ard.aget(self.add_prefix(key))
+        else:
+            value = await asyncio.to_thread(self._rd.get, self.add_prefix(key))
+        if value:
+            return loads(value)
+        return None
+
     def mget(self, keys):
         if not keys:
             return
@@ -46,22 +68,49 @@ def mget(self, keys):
             if value:
                 yield key, loads(value)
 
+    async def amget(self, keys):
+        if not keys:
+            return {}
+        prefixed_keys = [self.add_prefix(key) for key in keys]
+        if hasattr(self._ard, "amget"):
+            values = await self._ard.amget(prefixed_keys)
+        else:
+            values = await asyncio.to_thread(self._rd.mget, prefixed_keys)
+        return {key: loads(value) for key, value in zip(keys, values) if value}
+
     def set(self, key, value):
         old_value = self.get(key)
         self._rd.set(self.add_prefix(key), dumps(value))
         signals.config_updated.send(sender=config, key=key, old_value=old_value, new_value=value)
 
+    async def aset(self, key, value):
+        # We need the old value for the signal.
+        # Signals are synchronous in Django, but we can't easily change that here.
+        old_value = await self.aget(key)
+        if hasattr(self._ard, "aset"):
+            await self._ard.aset(self.add_prefix(key), dumps(value))
+        else:
+            await asyncio.to_thread(self._rd.set, self.add_prefix(key), dumps(value))
+        signals.config_updated.send(sender=config, key=key, old_value=old_value, new_value=value)
+
 
 class CachingRedisBackend(RedisBackend):
     _sentinel = object()
     _lock = RLock()
+    _async_lock = None  # Lazy-initialized asyncio.Lock
 
     def __init__(self):
         super().__init__()
         self._timeout = settings.REDIS_CACHE_TIMEOUT
         self._cache = {}
         self._sentinel = object()
 
+    def _get_async_lock(self):
+        # Lazily create the asyncio lock to avoid issues with event loops
+        if self._async_lock is None:
+            self._async_lock = asyncio.Lock()
+        return self._async_lock
+
     def _has_expired(self, value):
         return value[0] <= monotonic()
 
@@ -79,15 +128,70 @@ def get(self, key):
 
         return value[1]
 
+    async def aget(self, key):
+        value = self._cache.get(key, self._sentinel)
+
+        if value is self._sentinel or self._has_expired(value):
+            async with self._get_async_lock():
+                # Double-check after acquiring lock
+                value = self._cache.get(key, self._sentinel)
+                if value is self._sentinel or self._has_expired(value):
+                    new_value = await super().aget(key)
+                    self._cache_value(key, new_value)
+                    return new_value
+                return value[1]
+
+        return value[1]
+
     def set(self, key, value):
         with self._lock:
             super().set(key, value)
             self._cache_value(key, value)
 
+    async def aset(self, key, value):
+        async with self._get_async_lock():
+            await super().aset(key, value)
+            self._cache_value(key, value)
+
     def mget(self, keys):
         if not keys:
             return
         for key in keys:
             value = self.get(key)
             if value is not None:
                 yield key, value
+
+    async def amget(self, keys):
+        if not keys:
+            return {}
+
+        results = {}
+        missing_keys = []
+
+        # First, check the local cache for all keys
+        for key in keys:
+            value = self._cache.get(key, self._sentinel)
+            if value is not self._sentinel and not self._has_expired(value):
+                results[key] = value[1]
+            else:
+                missing_keys.append(key)
+
+        # Fetch missing keys from Redis
+        if missing_keys:
+            async with self._get_async_lock():
+                # Re-check cache for keys that might have been fetched while waiting for lock
+                still_missing = []
+                for key in missing_keys:
+                    value = self._cache.get(key, self._sentinel)
+                    if value is not self._sentinel and not self._has_expired(value):
+                        results[key] = value[1]
+                    else:
+                        still_missing.append(key)
+
+                if still_missing:
+                    fetched = await super().amget(still_missing)
+                    for key, value in fetched.items():
+                        self._cache_value(key, value)
+                        results[key] = value
+
+        return results