A Flexible Memcached Client for FastAPI

7 min read

Why Use This Memcached Client?

Caching can help your FastAPI application reduce repeated work and lower database load. By storing the results of expensive operations - like database queries or API calls - in memory, you can serve future requests more quickly and avoid unnecessary processing. This doesn't automatically make every part of your app faster, but it does help with endpoints or functions that fetch the same data often. Used thoughtfully, caching can improve response times and make your application more efficient. Most FastAPI caching libraries, like fastapi-cached2, use decorators to wrap route handlers. That approach is simple, but it can get in the way if you need more control or want to cache data outside of routes.

This Memcached client is built on top of aiomcache, a lightweight and fast asyncio Memcached library. By using aiomcache, you get efficient async operations and minimal overhead, making it a good fit for modern Python web applications.

Here’s what sets this client apart:

• Safe Memcached Key Generation: Memcached has strict rules for key length and allowed characters. This client automatically generates safe keys, hashes long keys, and replaces invalid characters, so you never have to worry about key errors.

• Flexible Key Parts: You can pass any number of strings, parameters, or tuples as key parts. The client will automatically join and sanitize them, making it easy to build descriptive and unique cache keys for any use case.

• Namespace Invalidation with Versioning: Memcached does not natively support namespace invalidation. This client implements it using a common versioning pattern: when you invalidate a namespace, the version is bumped and all previous keys become stale. This is a practical solution for cache invalidation at scale.

• No Decorators: You call the cache client directly, so you can cache data wherever you want, not just at the route level. When you use decorators like in fastapi-cached2, the request is intercepted, which can make it complex to log or monitor API requests in your application.

• Custom Key Generation: You can build keys from multiple parts and add prefixes for namespaces.

• Handles Encoding/Decoding: You don't need to worry about converting objects to strings or JSON - just use set_json and get_json.

• Easy FastAPI Dependency Injection: The client provides a simple dependency function, so you can inject it into your FastAPI routes with Depends.

Namespace invalidation is built in using versioning, so you can easily expire groups of related keys. The client also handles encoding and decoding for you, including JSON objects. If you need even faster serialization, you can swap in orjson later.

In short, this client is about flexibility and control. You decide what to cache, how to structure your keys, and when to invalidate or refresh cached data based on your application's needs.

The Code

Here's the full implementation:

import hashlib
import json
import logging
import re
from typing import Any, Optional

import aiomcache

MAX_KEY_LENGTH = 250
POOL_SIZE = 8
TTL_EXPIRE_SECONDS = 1000

def memcached_key(*parts, prefix=None) -> str:
    """
    Generate a memcached-safe key from given string parts.
    """
    key = ":".join(str(p) for p in parts)
    if prefix:
        key = f"{prefix}:{key}"
    key = re.sub(r"[^a-zA-Z0-9!#$%&'()*+\-./:<=>?@\[\]^_{|}~]", "_", key)
    key_bytes = key.encode("utf-8")
    if len(key_bytes) > MAX_KEY_LENGTH:
        hashed = hashlib.sha256(key_bytes).hexdigest()
        key = f"{key[:200]}:{hashed}"[:MAX_KEY_LENGTH]
    return key

class MemcachedClient:
    """
    A singleton async Memcached client for FastAPI applications using aiomcache.
    Handles all encoding/decoding internally.
    Supports namespace invalidation via versioning.
    """

    _instance: Optional["MemcachedClient"] = None

    def __init__(self, host: str = "127.0.0.1", port: int = 11211, pool_size: int = POOL_SIZE):
        self.host = host
        self.port = port
        self.pool_size = pool_size
        self.client: Optional[aiomcache.Client] = None

    @classmethod
    def get_instance(cls, host: str = "127.0.0.1", port: int = 11211):
        """Get the singleton instance of MemcachedClient."""
        if cls._instance is None:
            cls._instance = cls(host, port)
        return cls._instance

    async def init(self):
        """Initialize the aiomcache client if not already initialized."""
        if self.client is None:
            self.client = aiomcache.Client(self.host, self.port, pool_size=self.pool_size)

    # --- Namespace Versioning ---
    def _version_key(self, prefix: str) -> bytes:
        return f"__ns_version__:{prefix}".encode("utf-8")

    async def get_namespace_version(self, prefix: str) -> int:
        """Retrieve the current version for a namespace (prefix). Returns 1 if not set."""
        if self.client is None:
            await self.init()
        if self.client is None:
            return 1
        vkey = self._version_key(prefix)
        try:
            value = await self.client.get(vkey)
            return int(value) if value is not None else 1
        except (ValueError, OSError, aiomcache.exceptions.ClientException):
            return 1

    async def invalidate_namespace(self, prefix: str):
        """Invalidate a namespace by incrementing its version."""
        if self.client is None:
            await self.init()
        if self.client is None:
            return
        vkey = self._version_key(prefix)
        try:
            await self.client.incr(vkey, 1)
        except aiomcache.exceptions.ClientException:
            # Key not found, set initial version to 2 (since 1 is default)
            await self.client.set(vkey, b"2", exptime=0)
        except (OSError, aiomcache.exceptions.ClientException) as e:
            logging.error(f"Memcached invalidate_namespace error: {e!r}")

    # --- Core Key and Raw Operations ---
    async def _generate_versioned_key(self, *parts, prefix=None) -> bytes:
        """Generate a versioned memcached key asynchronously."""
        if prefix:
            version = await self.get_namespace_version(prefix)
            key_prefix = f"{prefix}:v{version}"
            key_str = memcached_key(*parts, prefix=key_prefix)
        else:
            key_str = memcached_key(*parts)
        return key_str.encode("utf-8")

    # --- Public String Key-Value Methods ---
    async def get(self, *key_parts, prefix=None) -> Optional[str]:
        """Retrieve a string value from memcached."""
        if self.client is None:
            await self.init()
        if self.client is None:
            return None
        key = await self._generate_versioned_key(*key_parts, prefix=prefix)
        try:
            value = await self.client.get(key)
            return value.decode("utf-8") if value is not None else None
        except (OSError, aiomcache.exceptions.ClientException) as e:
            logging.error(f"Memcached get error: {e!r}")
            return None

    async def set(self, value: str, *key_parts, prefix=None, expire: int = TTL_EXPIRE_SECONDS):
        """Set a string value in memcached."""
        if self.client is None:
            await self.init()
        if self.client is None:
            return
        key = await self._generate_versioned_key(*key_parts, prefix=prefix)
        try:
            await self.client.set(key, value.encode("utf-8"), exptime=expire)
        except (OSError, aiomcache.exceptions.ClientException) as e:
            logging.error(f"Memcached set error: {e!r}")

    # --- Public JSON Methods ---
    async def get_json(self, *key_parts, prefix=None) -> Optional[Any]:
        """Retrieve a JSON-encoded value and return as a Python object."""
        value_str = await self.get(*key_parts, prefix=prefix)
        if value_str:
            try:
                return json.loads(value_str)
            except json.JSONDecodeError:
                logging.warning(f"Failed to decode JSON for key '{key_parts}'")
                return None
        return None

    async def set_json(self, value: Any, *key_parts, prefix=None, expire: int = TTL_EXPIRE_SECONDS):
        """Encode a Python object as JSON and set it in memcached."""
        try:
            json_str = json.dumps(value)
            await self.set(json_str, *key_parts, prefix=prefix, expire=expire)
        except (TypeError, Exception) as e:
            logging.error(f"Memcached set_json error: {e!r}")

    # --- Deletion and Utility Methods ---
    async def delete(self, *key_parts, prefix=None):
        """Delete a key from memcached."""
        if self.client is None:
            await self.init()
        if self.client is None:
            return
        key = await self._generate_versioned_key(*key_parts, prefix=prefix)
        try:
            await self.client.delete(key)
        except (OSError, aiomcache.exceptions.ClientException) as e:
            logging.error(f"Memcached delete error: {e!r}")

    async def close(self):
        """Gracefully close the memcached client connection."""
        if self.client is not None:
            await self.client.close()
            self.client = None

    async def flush_all(self):
        """Delete everything in memcached. Use with caution."""
        if self.client is None:
            await self.init()
        if self.client is None:
            return
        try:
            await self.client.flush_all()
        except (OSError, aiomcache.exceptions.ClientException) as e:
            logging.error(f"Memcached flush_all error: {e!r}")

# FastAPI dependency
def get_memcached_client():
    return MemcachedClient.get_instance()

Example Usage in FastAPI

Here's how you might use this client in your FastAPI app:

from fastapi import FastAPI, Depends
from app.memcache_client import get_memcached_client

app = FastAPI()

@app.get("/user/{user_id}")
async def get_user(user_id: int, cache=Depends(get_memcached_client)):
    # Try to get user data from cache
    cached = await cache.get_json("user", user_id)
    if cached:
        return {"cached": True, "user": cached}

    # Otherwise, fetch from database (pseudo-code)
    user = {"id": user_id, "name": "Alice"}
    await cache.set_json(user, "user", user_id)
    return {"cached": False, "user": user}

Example Usage with Namespace/Tag/Prefix and TTL

You can also use a prefix/namespace/tag (like "user_profile") and set a custom TTL for cache expiration:

async def get_cached_user_profile(user_id: int, cache: MemcachedClient):
    # Try to get the user profile from cache with a namespace
    cached_profile = await cache.get_json(user_id, prefix="user_profile")
    if cached_profile:
        return cached_profile

    # Otherwise, fetch from database or external API (pseudo-code)
    profile = {"id": user_id, "bio": "This is a user profile."}
    # Set cache with a custom TTL (e.g., 600 seconds)
    await cache.set_json(profile, user_id, prefix="user_profile", expire=600)
    return profile

Note: The TTL (time-to-live) setting is in seconds and is optional.

Example Usage in a Utility Function

You can also use the Memcached client in utility functions, not just in FastAPI routes. Just pass an instance of the client to the function and use the caching methods as needed.

async def get_cached_user_profile(user_id: int, cache: MemcachedClient):
    # Try to get the user profile from cache
    cached_profile = await cache.get_json("user_profile", user_id)
    if cached_profile:
        return cached_profile

    # Otherwise, fetch from database or external API (pseudo-code)
    profile = {"id": user_id, "bio": "This is a user profile."}
    await cache.set_json(profile, "user_profile", user_id)
    return profile

How Is This Different from fastapi_cached2?

• Support for Namespaces/Tags as Prefixes: You can organize your cached data using prefixes, namespaces, or tags. This makes it easy to group related keys. As mentioned above, namespaces are not natively supported by memcached. See namespace invalidation below for more details.

• Supports Namespace invalidation: You can invalidate a whole group of cached keys by bumping the namespace version. Example:

  await cache.invalidate_namespace("user_profile")
  

• Custom and Flexible keys with key parts: You can create keys from any combination of parameters or strings. You can pass any combination of strings, parameters, or tuples as key parts. For example:

  await cache.get_json("order", order_id, "details")
  await cache.set_json(order_data, "order", order_id, "details")
  

• Handles encoding/decoding: You don't need to worry about converting objects to strings or JSON - just use set_json and get_json.

• Easy dependency injection: Just use Depends(get_memcached_client) in your FastAPI routes.

• No decorators: You call the cache client directly, so you can cache data wherever you want, not just at the API route level.

FAQ

Why not use a decorator for caching? Decorators are fine for simple cases, but they can get in the way if you need more control. This client lets you cache data wherever you want, not just at the route level.

Can I cache complex Python objects? Yes, as long as they can be serialized to JSON. For even faster json serialization, you could swap out json for orjson in the future.

How do I invalidate a group of cached keys? Use the invalidate_namespace method with a prefix. This bumps the version for that namespace, so all old keys become stale.

Is this client thread-safe? It uses a singleton pattern and async methods, so it's safe for typical FastAPI usage.

Can I use this with other frameworks? The client is designed for FastAPI, but you could adapt it for other async Python frameworks.

This Memcached client gives you more flexibility and control over caching in your FastAPI app. If you need custom key handling, namespace invalidation, or just want to avoid decorator-based caching, give it a try in your next project.