Redis Cache and Session Internals

Session backend

codex_django.sessions.backends.redis

Redis-backed SessionBase implementation.

Wire this up as SESSION_ENGINE = "codex_django.sessions.backends.redis".

Storage contract:

• Session payload is stored as the Django encoded string produced by :py:meth:SessionBase.encode. No pickle is used — the project must rely on a JSON-based SESSION_SERIALIZER (Django's default django.contrib.sessions.serializers.JSONSerializer is fine).
• Redis key format: {PROJECT_NAME}:{CODEX_SESSION_KEY_PREFIX}:{session_key}.
• TTL equals self.get_expiry_age() (driven by SESSION_COOKIE_AGE / per-session overrides).
• Redis failures propagate as :py:class:codex_platform.redis_service.exceptions.RedisConnectionError (or RedisServiceError). They are not swallowed — a broken Redis must never look like a valid empty session.

Legacy django-redis / pickled sessions are not readable by this backend and do not need to be migrated — old keys are simply ignored and the user re-authenticates.

Classes

SessionStore

Bases: SessionBase

Session store that persists session payloads in Redis as strings.

Instances are cheap and stateless with respect to Redis connections — the manager builds clients lazily.

Source code in src/codex_django/sessions/backends/redis.py

class SessionStore(SessionBase):
    """Session store that persists session payloads in Redis as strings.

    Instances are cheap and stateless with respect to Redis connections — the
    manager builds clients lazily.
    """

    def __init__(self, session_key: str | None = None) -> None:
        super().__init__(session_key)
        self._manager: Any = None

    # ---- infrastructure ---------------------------------------------------

    @property
    def redis_manager(self) -> Any:
        if self._manager is None:
            self._manager = get_default_redis_manager()
        return self._manager

    @staticmethod
    def _key_prefix() -> str:
        return str(getattr(settings, "CODEX_SESSION_KEY_PREFIX", _DEFAULT_PREFIX))

    def _key(self, session_key: str) -> str:
        # Build the namespaced key without an extra make_key call since
        # we prefix manually to match old behaviour, or use the manager's make_key
        # but old code used namespaced_key(self._key_prefix(), session_key)
        # We can just use the manager's make_key by passing prefix to manager
        # or doing it inline.
        parts = [getattr(settings, "PROJECT_NAME", ""), self._key_prefix(), session_key]
        return ":".join(p for p in parts if p)

    # ---- async core -------------------------------------------------------

    async def aexists(self, session_key: str) -> bool:
        async with self.redis_manager.async_string() as s:
            return bool(await s.exists(self._key(session_key)))

    async def acreate(self) -> None:
        aget_new_session_key = cast(Callable[[], Awaitable[str]], self._aget_new_session_key)  # type: ignore[attr-defined]
        while True:
            self._session_key = await aget_new_session_key()
            try:
                await self.asave(must_create=True)
            except CreateError:
                continue
            self.modified = True
            return

    async def aload(self) -> dict[str, Any]:
        key = self.session_key
        if key is None:
            return {}
        async with self.redis_manager.async_string() as s:
            raw = await s.get(self._key(key))
        if raw is None:
            return {}
        # SessionBase.decode returns {} for tampered/unreadable payloads.
        return dict(self.decode(raw))

    async def asave(self, must_create: bool = False) -> None:
        if self.session_key is None:
            await self.acreate()
            return

        aget_session = cast(Callable[[bool], Awaitable[dict[str, Any]]], self._aget_session)  # type: ignore[attr-defined]
        payload = self.encode(await aget_session(must_create))
        ttl = self.get_expiry_age()
        key = self._key(self.session_key)

        async with self.redis_manager.async_string() as s:
            if must_create:
                created = await s.client.set(key, payload, nx=True, ex=ttl)
                if not created:
                    raise CreateError
                return

            if not await s.exists(key):
                raise UpdateError
            await s.set(key, payload, ttl=ttl)

    async def adelete(self, session_key: str | None = None) -> None:
        if session_key is None:
            if self.session_key is None:
                return
            session_key = self.session_key
        async with self.redis_manager.async_string() as s:
            await s.delete(self._key(session_key))

    # ---- sync wrappers ----------------------------------------------------

    def exists(self, session_key: str) -> bool:
        with self.redis_manager.sync_string() as s:
            return bool(s.exists(self._key(session_key)))

    def create(self) -> None:
        get_new_session_key = cast(Callable[[], str], self._get_new_session_key)  # type: ignore[attr-defined]
        while True:
            self._session_key = get_new_session_key()
            try:
                self.save(must_create=True)
            except CreateError:
                continue
            self.modified = True
            return

    def load(self) -> dict[str, Any]:
        key = self.session_key
        if key is None:
            return {}
        with self.redis_manager.sync_string() as s:
            raw = s.get(self._key(key))
        if raw is None:
            return {}
        return dict(self.decode(raw))

    def save(self, must_create: bool = False) -> None:
        if self.session_key is None:
            self.create()
            return

        get_session = cast(Callable[[bool], dict[str, Any]], self._get_session)  # type: ignore[attr-defined]
        payload = self.encode(get_session(must_create))
        ttl = self.get_expiry_age()
        key = self._key(self.session_key)

        with self.redis_manager.sync_string() as s:
            if must_create:
                created = s.client.set(key, payload, nx=True, ex=ttl)
                if not created:
                    raise CreateError
                return

            if not s.exists(key):
                raise UpdateError
            s.set(key, payload, ttl=ttl)

    def delete(self, session_key: str | None = None) -> None:
        if session_key is None:
            if self.session_key is None:
                return
            session_key = self.session_key
        with self.redis_manager.sync_string() as s:
            s.delete(self._key(session_key))

    # ---- misc -------------------------------------------------------------

    @classmethod
    def clear_expired(cls) -> None:
        """Redis expires keys on its own — nothing to sweep."""
        return None

    @classmethod
    async def aclear_expired(cls) -> None:
        return None

Functions

clear_expired() classmethod

Redis expires keys on its own — nothing to sweep.

Source code in src/codex_django/sessions/backends/redis.py

@classmethod
def clear_expired(cls) -> None:
    """Redis expires keys on its own — nothing to sweep."""
    return None

Functions

Cache backend

codex_django.cache.backends.redis

Redis-backed Django cache backend (JSON, no pickle).

Wire this up as::

CACHES = {
    "default": {
        "BACKEND": "codex_django.cache.backends.redis.RedisCache",
        "LOCATION": REDIS_URL,       # optional; falls back to settings.REDIS_URL
        "KEY_PREFIX": PROJECT_NAME,  # required for clear() to be namespace-scoped
        "TIMEOUT": 300,
        "OPTIONS": {
            # Optional: "SERIALIZER": "my_app.cache.CustomSerializer",
        },
    }
}

Design highlights:

• Values are stored as UTF-8 JSON strings (see :mod:codex_django.cache.serializers). Non-JSON-native values raise TypeError; there is no pickle fallback.
• add uses Redis SET NX EX atomically.
• clear uses SCAN + DEL scoped to {key_prefix}:*. If KEY_PREFIX is empty, clear refuses to run — never FLUSHDB.
• Redis errors propagate to the caller as :class:codex_platform.redis_service.exceptions.RedisConnectionError / RedisServiceError — the cache does not silently degrade.

Classes

RedisCache

Bases: BaseCache

Django cache backend that stores JSON strings in Redis.

Source code in src/codex_django/cache/backends/redis.py

class RedisCache(BaseCache):
    """Django cache backend that stores JSON strings in Redis."""

    def __init__(self, server: str | None, params: dict[str, Any]) -> None:
        super().__init__(params)
        self._location = str(server or getattr(settings, "REDIS_URL", "redis://localhost:6379/0"))
        options = params.get("OPTIONS") or {}
        serializer_path = options.get("SERIALIZER")
        self._serializer: Serializer = _import_serializer(serializer_path) if serializer_path else JsonSerializer()
        self._manager: BaseDjangoRedisManager | None = None

    # ---- infrastructure ---------------------------------------------------

    @property
    def redis_manager(self) -> BaseDjangoRedisManager:
        if self._manager is None:

            def async_factory() -> AsyncRedis:
                return cast(AsyncRedis, AsyncRedis.from_url(self._location, decode_responses=True))

            def sync_factory() -> SyncRedis:
                return SyncRedis.from_url(self._location, decode_responses=True)

            self._manager = BaseDjangoRedisManager(
                prefix=self.key_prefix,
                async_client_factory=async_factory,
                sync_client_factory=sync_factory,
            )
        return self._manager

    def _resolve_ttl(self, timeout: Any) -> int | None | Any:
        """Return ``None`` (persist), ``0`` (delete), or positive seconds.

        ``DEFAULT_TIMEOUT`` (the sentinel Django passes when the caller did
        not specify one) is replaced by ``self.default_timeout``.
        """
        if timeout is DEFAULT_TIMEOUT:
            timeout = self.default_timeout
        if timeout is None:
            return None
        try:
            ttl = int(timeout)
        except (TypeError, ValueError):
            return None
        if ttl <= 0:
            return 0
        return ttl

    def _dump(self, value: Any) -> str:
        return self._serializer.dumps(value)

    def _load(self, raw: str | None, default: Any) -> Any:
        if raw is None:
            return default
        return self._serializer.loads(raw)

    # ---- async core -------------------------------------------------------

    async def aget(self, key: Any, default: Any = None, version: Any = None) -> Any:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        async with self.redis_manager.async_string() as s:
            raw = await s.get(real_key)
        return self._load(raw, default)

    async def aset(
        self,
        key: Any,
        value: Any,
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> None:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        ttl = self._resolve_ttl(timeout)
        if ttl == 0:
            async with self.redis_manager.async_string() as s:
                await s.delete(real_key)
            return
        payload = self._dump(value)
        async with self.redis_manager.async_string() as s:
            await s.set(real_key, payload, ttl=ttl)

    async def aadd(
        self,
        key: Any,
        value: Any,
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        ttl = self._resolve_ttl(timeout)
        if ttl == 0:
            return False
        payload = self._dump(value)
        async with self.redis_manager.async_string() as s:
            # Use raw client for setnx + ex since operations wrapper might not support atomic setnx with ex
            # Actually StringOperations set_nx doesn't take ttl in some versions, or maybe it uses set(nx=True)
            # We'll use the raw client exposed by operation or directly if not available.
            return bool(await s.client.set(real_key, payload, nx=True, ex=ttl))

    async def adelete(self, key: Any, version: Any = None) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        async with self.redis_manager.async_string() as s:
            existed = await s.exists(real_key)
            if not existed:
                return False
            await s.delete(real_key)
        return True

    async def ahas_key(self, key: Any, version: Any = None) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        async with self.redis_manager.async_string() as s:
            return bool(await s.exists(real_key))

    async def atouch(
        self,
        key: Any,
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        ttl = self._resolve_ttl(timeout)
        async with self.redis_manager.async_string() as s:
            if ttl == 0:
                existed = await s.exists(real_key)
                if existed:
                    await s.delete(real_key)
                return bool(existed)
            if ttl is None:
                return bool(await s.client.persist(real_key))
            return bool(await s.expire(real_key, ttl))

    async def aget_many(self, keys: Iterable[Any], version: Any = None) -> dict[Any, Any]:
        key_list = list(keys)
        if not key_list:
            return {}
        real_keys = [self.make_key(k, version=version) for k in key_list]
        for real in real_keys:
            self.validate_key(real)
        async with self.redis_manager.async_string() as s:
            raws = await s.mget(*real_keys)
        result: dict[Any, Any] = {}
        for original, raw in zip(key_list, raws, strict=True):
            if raw is None:
                continue
            result[original] = self._serializer.loads(raw)
        return result

    async def aset_many(
        self,
        mapping: dict[Any, Any],
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> list[Any]:
        if not mapping:
            return []
        for original in mapping:
            await self.aset(original, mapping[original], timeout=timeout, version=version)
        return []

    async def adelete_many(self, keys: Iterable[Any], version: Any = None) -> None:
        for key in keys:
            await self.adelete(key, version=version)

    async def aincr(self, key: Any, delta: int = 1, version: Any = None) -> int:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        async with self.redis_manager.async_string() as s:
            if not await s.exists(real_key):
                raise ValueError(f"Key '{key}' not found")
            return int(await s.incr(real_key, delta))

    async def adecr(self, key: Any, delta: int = 1, version: Any = None) -> int:
        return await self.aincr(key, -delta, version=version)

    async def aclear(self) -> None:
        if not self.key_prefix:
            raise ImproperlyConfigured(
                "codex_django RedisCache.clear() requires CACHES['default']['KEY_PREFIX'] "
                "to be set; refusing to run an unbounded SCAN+DEL."
            )
        async with self.redis_manager.async_string() as s:
            # We'll collect and delete manually or use raw client's eval if needed,
            # but usually async clients have scan_iter
            cursor = 0
            pattern = f"{self.key_prefix}:*"
            while True:
                cursor, keys = await s.client.scan(cursor=cursor, match=pattern, count=100)
                if keys:
                    await s.client.delete(*keys)
                if cursor == 0:
                    break

    # ---- sync wrappers ----------------------------------------------------

    def get(self, key: Any, default: Any = None, version: Any = None) -> Any:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        with self.redis_manager.sync_string() as s:
            raw = s.get(real_key)
        return self._load(raw, default)

    def set(
        self,
        key: Any,
        value: Any,
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> None:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        ttl = self._resolve_ttl(timeout)
        if ttl == 0:
            with self.redis_manager.sync_string() as s:
                s.delete(real_key)
            return
        payload = self._dump(value)
        with self.redis_manager.sync_string() as s:
            s.set(real_key, payload, ttl=ttl)

    def add(
        self,
        key: Any,
        value: Any,
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        ttl = self._resolve_ttl(timeout)
        if ttl == 0:
            return False
        payload = self._dump(value)
        with self.redis_manager.sync_string() as s:
            return bool(s.client.set(real_key, payload, nx=True, ex=ttl))

    def delete(self, key: Any, version: Any = None) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        with self.redis_manager.sync_string() as s:
            existed = s.exists(real_key)
            if not existed:
                return False
            s.delete(real_key)
        return True

    def has_key(self, key: Any, version: Any = None) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        with self.redis_manager.sync_string() as s:
            return bool(s.exists(real_key))

    def touch(
        self,
        key: Any,
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> bool:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        ttl = self._resolve_ttl(timeout)
        with self.redis_manager.sync_string() as s:
            if ttl == 0:
                existed = s.exists(real_key)
                if existed:
                    s.delete(real_key)
                return bool(existed)
            if ttl is None:
                return bool(s.client.persist(real_key))
            return bool(s.expire(real_key, ttl))

    def get_many(self, keys: Iterable[Any], version: Any = None) -> dict[Any, Any]:
        key_list = list(keys)
        if not key_list:
            return {}
        real_keys = [self.make_key(k, version=version) for k in key_list]
        for real in real_keys:
            self.validate_key(real)
        with self.redis_manager.sync_string() as s:
            raws = s.mget(*real_keys)
        result: dict[Any, Any] = {}
        for original, raw in zip(key_list, raws, strict=True):
            if raw is None:
                continue
            result[original] = self._serializer.loads(raw)
        return result

    def set_many(
        self,
        mapping: dict[Any, Any],
        timeout: Any = DEFAULT_TIMEOUT,
        version: Any = None,
    ) -> list[Any]:
        if not mapping:
            return []
        for original in mapping:
            self.set(original, mapping[original], timeout=timeout, version=version)
        return []

    def delete_many(self, keys: Iterable[Any], version: Any = None) -> None:
        for key in keys:
            self.delete(key, version=version)

    def incr(self, key: Any, delta: int = 1, version: Any = None) -> int:
        real_key = self.make_key(key, version=version)
        self.validate_key(real_key)
        with self.redis_manager.sync_string() as s:
            if not s.exists(real_key):
                raise ValueError(f"Key '{key}' not found")
            return int(s.incr(real_key, delta))

    def decr(self, key: Any, delta: int = 1, version: Any = None) -> int:
        return self.incr(key, -delta, version=version)

    def clear(self) -> None:
        if not self.key_prefix:
            raise ImproperlyConfigured(
                "codex_django RedisCache.clear() requires CACHES['default']['KEY_PREFIX'] "
                "to be set; refusing to run an unbounded SCAN+DEL."
            )
        with self.redis_manager.sync_string() as s:
            client: Any = s.client
            cursor = 0
            pattern = f"{self.key_prefix}:*"
            while True:
                cursor, keys = client.scan(cursor=cursor, match=pattern, count=100)
                if keys:
                    client.delete(*keys)
                if cursor == 0:
                    break

JSON serializer

codex_django.cache.serializers

Serializers for the Redis cache backend.

Default policy: strict JSON. Any value that cannot be encoded as JSON raises :class:TypeError — the backend never silently falls back to pickle. Callers that need to cache non-JSON-native values (datetime, Decimal, UUID, set …) should use the helpers in :mod:codex_django.cache.values to convert them explicitly.

A downstream project may swap the serializer via OPTIONS["SERIALIZER"] (dotted path to a class implementing :class:Serializer).

Classes

Serializer

Bases: Protocol

Minimal cache serializer interface.

Source code in src/codex_django/cache/serializers.py

class Serializer(Protocol):
    """Minimal cache serializer interface."""

    def dumps(self, value: Any) -> str:  # pragma: no cover - protocol
        ...

    def loads(self, raw: str) -> Any:  # pragma: no cover - protocol
        ...

JsonSerializer

Strict JSON serializer — UTF-8, no ensure_ascii escaping.

Source code in src/codex_django/cache/serializers.py

class JsonSerializer:
    """Strict JSON serializer — UTF-8, no ``ensure_ascii`` escaping."""

    def dumps(self, value: Any) -> str:
        return json.dumps(value, ensure_ascii=False, separators=(",", ":"))

    def loads(self, raw: str) -> Any:
        return json.loads(raw)

CacheCoder

codex_django.cache.values

Explicit helpers for caching non-JSON-native Python values.

The Redis cache backend stores values as strict JSON and raises TypeError for anything that json.dumps cannot handle (datetime, date, Decimal, UUID, set …). Rather than smuggle magic type tags into the serializer (which would leak into every Redis consumer and make reverting impossible), we expose an explicit coder that callers invoke themselves::

from codex_django.cache.values import CacheCoder

cache.set("user:42:last_seen", CacheCoder.dump_datetime(user.last_seen), 3600)
raw = cache.get("user:42:last_seen")
last_seen = CacheCoder.load_datetime(raw) if raw else None

The dump() / load() helpers cover ad-hoc payloads (nested dict / list / tuple); load() requires explicit hints because JSON does not preserve Python types.

Classes

CacheCoder

Round-trip helpers for non-JSON-native values.

Every helper is a pure function — no Redis access, no Django settings lookup — so the class is importable and usable anywhere.

Source code in src/codex_django/cache/values.py

class CacheCoder:
    """Round-trip helpers for non-JSON-native values.

    Every helper is a pure function — no Redis access, no Django settings
    lookup — so the class is importable and usable anywhere.
    """

    # ---- datetime ------------------------------------------------------

    @staticmethod
    def dump_datetime(value: datetime) -> str:
        return value.isoformat()

    @staticmethod
    def load_datetime(raw: str) -> datetime:
        return datetime.fromisoformat(raw)

    # ---- date ----------------------------------------------------------

    @staticmethod
    def dump_date(value: date) -> str:
        return value.isoformat()

    @staticmethod
    def load_date(raw: str) -> date:
        return date.fromisoformat(raw)

    # ---- time ----------------------------------------------------------

    @staticmethod
    def dump_time(value: time) -> str:
        return value.isoformat()

    @staticmethod
    def load_time(raw: str) -> time:
        return time.fromisoformat(raw)

    # ---- timedelta -----------------------------------------------------

    @staticmethod
    def dump_timedelta(value: timedelta) -> float:
        return value.total_seconds()

    @staticmethod
    def load_timedelta(raw: float | int) -> timedelta:
        return timedelta(seconds=float(raw))

    # ---- decimal -------------------------------------------------------

    @staticmethod
    def dump_decimal(value: Decimal) -> str:
        return str(value)

    @staticmethod
    def load_decimal(raw: str) -> Decimal:
        return Decimal(raw)

    # ---- uuid ----------------------------------------------------------

    @staticmethod
    def dump_uuid(value: UUID) -> str:
        return str(value)

    @staticmethod
    def load_uuid(raw: str) -> UUID:
        return UUID(raw)

    # ---- set -----------------------------------------------------------

    @staticmethod
    def dump_set(value: set[Any] | frozenset[Any]) -> list[Any]:
        return list(value)

    @staticmethod
    def load_set(raw: Sequence[Any]) -> set[Any]:
        return set(raw)

    # ---- bytes ---------------------------------------------------------

    @staticmethod
    def dump_bytes(value: bytes) -> str:
        """Encode ``bytes`` as a hex string (roundtrip-safe, ASCII)."""
        return value.hex()

    @staticmethod
    def load_bytes(raw: str) -> bytes:
        return bytes.fromhex(raw)

    # ---- scalar helpers ------------------------------------------------

    @staticmethod
    def dump_bool(value: bool) -> str:
        return "1" if value else "0"

    @staticmethod
    def load_bool(raw: str) -> bool:
        return raw == "1"

    @staticmethod
    def dump_none() -> str:
        return ""

    # ---- generic recursive dump ---------------------------------------

    @classmethod
    def dump(cls, value: Any) -> Any:
        """Best-effort conversion of a nested structure to JSON-native types.

        Recurses through ``dict`` / ``list`` / ``tuple``. Any value of an
        unknown type is returned as-is — JSON serialization will raise
        ``TypeError`` on it later, which is the intended strict-mode signal.
        """
        if isinstance(value, bool):
            return cls.dump_bool(value)
        if value is None:
            return cls.dump_none()
        if isinstance(value, str | int | float):
            return value
        if isinstance(value, Enum):
            return cls.dump(value.value)
        if isinstance(value, Promise):
            return force_str(value)
        if isinstance(value, Path):
            return str(value)
        if isinstance(value, datetime):
            return cls.dump_datetime(value)
        if isinstance(value, date):
            return cls.dump_date(value)
        if isinstance(value, time):
            return cls.dump_time(value)
        if isinstance(value, timedelta):
            return cls.dump_timedelta(value)
        if isinstance(value, Decimal):
            return cls.dump_decimal(value)
        if isinstance(value, UUID):
            return cls.dump_uuid(value)
        if isinstance(value, set | frozenset):
            return [cls.dump(v) for v in value]
        if isinstance(value, bytes):
            return cls.dump_bytes(value)
        if isinstance(value, Mapping):
            return {k: cls.dump(v) for k, v in value.items()}
        if isinstance(value, list | tuple):
            return [cls.dump(v) for v in value]
        return value

Functions

dump_bytes(value) staticmethod

Encode bytes as a hex string (roundtrip-safe, ASCII).

Source code in src/codex_django/cache/values.py

@staticmethod
def dump_bytes(value: bytes) -> str:
    """Encode ``bytes`` as a hex string (roundtrip-safe, ASCII)."""
    return value.hex()

dump(value) classmethod

Best-effort conversion of a nested structure to JSON-native types.

Recurses through dict / list / tuple. Any value of an unknown type is returned as-is — JSON serialization will raise TypeError on it later, which is the intended strict-mode signal.

Source code in src/codex_django/cache/values.py

@classmethod
def dump(cls, value: Any) -> Any:
    """Best-effort conversion of a nested structure to JSON-native types.

    Recurses through ``dict`` / ``list`` / ``tuple``. Any value of an
    unknown type is returned as-is — JSON serialization will raise
    ``TypeError`` on it later, which is the intended strict-mode signal.
    """
    if isinstance(value, bool):
        return cls.dump_bool(value)
    if value is None:
        return cls.dump_none()
    if isinstance(value, str | int | float):
        return value
    if isinstance(value, Enum):
        return cls.dump(value.value)
    if isinstance(value, Promise):
        return force_str(value)
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, datetime):
        return cls.dump_datetime(value)
    if isinstance(value, date):
        return cls.dump_date(value)
    if isinstance(value, time):
        return cls.dump_time(value)
    if isinstance(value, timedelta):
        return cls.dump_timedelta(value)
    if isinstance(value, Decimal):
        return cls.dump_decimal(value)
    if isinstance(value, UUID):
        return cls.dump_uuid(value)
    if isinstance(value, set | frozenset):
        return [cls.dump(v) for v in value]
    if isinstance(value, bytes):
        return cls.dump_bytes(value)
    if isinstance(value, Mapping):
        return {k: cls.dump(v) for k, v in value.items()}
    if isinstance(value, list | tuple):
        return [cls.dump(v) for v in value]
    return value

Shared adapter

codex_django.core.redis.django_adapter

Shared Redis builders for Django session and cache backends.

Provides the minimal surface needed by codex_django.sessions and codex_django.cache without going through BaseDjangoRedisManager.

We intentionally do NOT reuse BaseDjangoRedisManager here because session/cache backends must fail loud on Redis outage; the _is_disabled shortcut (used by domain managers in DEBUG mode) is wrong for them.

Functions

build_redis_client(url=None)

Return an async Redis client with decode_responses=True.

Strings are always returned as str (not bytes) to match the behavior relied upon by Django's session/cache layers.

Source code in src/codex_django/core/redis/django_adapter.py

def build_redis_client(url: str | None = None) -> Redis:
    """Return an async Redis client with ``decode_responses=True``.

    Strings are always returned as ``str`` (not ``bytes``) to match the
    behavior relied upon by Django's session/cache layers.
    """
    client: Redis = Redis.from_url(_resolve_url(url), decode_responses=True)
    return client

build_redis_service(url=None)

Return a RedisService bound to an async Redis client.

Source code in src/codex_django/core/redis/django_adapter.py

def build_redis_service(url: str | None = None) -> RedisService:
    """Return a ``RedisService`` bound to an async Redis client."""
    return RedisService(build_redis_client(url))

namespaced_key(prefix, key, *, project_name=None)

Build a colon-delimited Redis key: {PROJECT_NAME}:{prefix}:{key}.

Empty segments are omitted. Mirrors BaseDjangoRedisManager.make_key.

Source code in src/codex_django/core/redis/django_adapter.py

def namespaced_key(prefix: str, key: str, *, project_name: str | None = None) -> str:
    """Build a colon-delimited Redis key: ``{PROJECT_NAME}:{prefix}:{key}``.

    Empty segments are omitted. Mirrors ``BaseDjangoRedisManager.make_key``.
    """
    if project_name is None:
        project_name = str(getattr(settings, "PROJECT_NAME", ""))
    parts = [p for p in (project_name, prefix, key) if p]
    return ":".join(parts)