# Implementing a Per-Request Cache Decorator in Django

Our goal is to implement a `per_request_cache` decorator in Django that can cache function results on a per-request basis. Here's what we want to achieve:

@per_request_cache
def expensive_calculation(param):
    # Some expensive operation
    return result

This decorator should cache the result of `expensive_calculation` for each unique parameter, but only for the duration of the current request.

## The Challenge

Implementing this decorator presents several challenges:

1. How to associate cache data with specific requests
2. How to access the current request within a decorator
3. How to handle concurrent requests
4. How to manage cache data in multi-threaded environments

## Initial Implementation

Let's start with a basic implementation of our `per_request_cache` decorator:

caches = {}

class per_request_cache:
    def __init__(self, fn):
        self._fn = fn

    @property
    def _cache(self):
        return caches[current_request].setdefault(self, {})

    def __call__(self, arg):
        if arg not in self._cache:
            self._cache[arg] = self._fn(arg)
        return self._cache[arg]

This implementation looks promising, but it relies on a `current_request` variable that we haven't defined yet.

## The Global Variable Pitfall

One might consider storing the current request in a global variable:

current_request = None

class RequestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        global current_request
        current_request = request
        response = self.get_response(request)
        return response

However, this approach fails when dealing with concurrent requests. If request 2 is invoked before request 1 finishes, any subsequent calculations for request 1 will incorrectly use request 2's context.

## Thread-Local Storage: A Step Forward

To handle concurrent requests, we can use thread-local storage:

import threading

_local = threading.local()

class RequestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        _local.request = request
        response = self.get_response(request)
        return response

Thread-local allows each thread to have its own isolated storage. Here's how it works:

1. An empty dictionary is initialized when a `threading.local()` object is created.
2. The current thread's ID is used as a key to fetch a thread-specific dictionary whenever you access an attribute on this object (such as `_local.request`).
3. Using the attribute name in this thread-specific dictionary, it retrieves or sets the object attribute.

This means that even if multiple threads are accessing `_local.request` simultaneously, they're actually accessing `request` objects from different dictionaries, each specific to their own thread. This isolation prevents data races and ensures that each thread (and thus each request) has its own separate storage.

Now we can update our `per_request_cache` to use `_local.request`:

@property
def _cache(self):
    return caches[_local.request].setdefault(self, {})

## Managing Cache Lifecycle

We need to ensure that the cache is cleared after each request. We can do this with a middleware:

class ClearCacheMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        _local.request = request
        caches[request] = {}
        response = self.get_response(request)
        caches.pop(request)
        return response

## The Multi-Threading Dilemma

Thread-local storage solves our concurrency issue but introduces a new problem. If we create new threads within a request to perform parallel tasks, these new threads won't have access to the current request.

from concurrent.futures import ThreadPoolExecutor

@per_request_cache
def expensive_calculation(arg):
    # Some expensive operation
    return result

def multiple_expensive_calculation(args):
    with ThreadPoolExecutor() as executor:
        results = list(executor.map(expensive_calculation, args))
    return results

Here, `ThreadPoolExecutor` creates new threads to execute tasks concurrently. However, these worker threads don't inherit the thread-local data from the parent thread, leaving them without access to the current request.

## The Solution: Thread Initialization

To solve this, we need to pass the current request to each new thread we create:

def set_request(request):
    _local.request = request

def get_thread_pool(max_workers=None, thread_name_prefix=""):
    return ThreadPoolExecutor(
        max_workers=max_workers,
        thread_name_prefix=thread_name_prefix,
        initializer=set_request,
        initargs=(_local.request,)
    )

The `ThreadPoolExecutor` constructor's `initializer` and `initargs` parameters are crucial for our solution:

- `initializer`: a function executed in each new worker thread.
- `initargs`: arguments for the initializer, prepared in the current thread.

This allows data passage between threads:

- `set_request` is our `initializer` that sets `_local.request` in new threads.
- `(_local.request,)` is our `initargs`. It captures the current request from the original thread, which will be passed to `set_request`.

This ensures all executor threads access the correct request, even though they're running in different threads from the original request. Now, instead of creating a `ThreadPoolExecutor` directly, we use our `get_thread_pool` function:

def multiple_expensive_calculation(args):
    with get_thread_pool() as executor:
        results = list(executor.map(expensive_calculation, args))
    return results

This ensures that all threads created by the executor have access to the current request, solving our multi-threading dilemma.

## Putting It All Together

Here's our complete implementation of a per-request cache decorator:

from __future__ import annotations

import threading
from collections.abc import Callable, Hashable
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from typing import Annotated, Any, Generic, ParamSpec, TypeVar

from django.http import HttpRequest
from django.http.response import HttpResponse

CacheContext = Annotated[Hashable, "CacheContext"]
CacheScope = Annotated[Hashable, "CacheScope"]
CacheKey = Annotated[Hashable, "CacheKey"]

P = ParamSpec("P")
R = TypeVar("R")

NO_CACHE = object()


class _ThreadLocal(threading.local):
    ctx: CacheContext = NO_CACHE


_local = _ThreadLocal()
caches: dict[CacheContext, dict[CacheScope, dict[CacheKey, Any]]] = {}


@contextmanager
def cache_context(ctx: CacheContext) -> Any:
    _local.ctx = ctx
    caches[ctx] = {}

    yield

    caches.pop(ctx)
    _local.ctx = NO_CACHE


class ClearCacheMiddleware:
    """Clears local cache after each request"""

    def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
        self.get_response = get_response

    def __call__(self, request: HttpRequest) -> HttpResponse:
        with cache_context(request):
            return self.get_response(request)


class per_reqeust_cache(Generic[P, R]):
    def __init__(self, fn: Callable[P, R]) -> None:
        self._fn = fn

    def _cache(self) -> dict[Any, Any]:
        if _local.ctx is NO_CACHE:
            return {}

        return caches[_local.ctx].setdefault(self, {})

    def __call__(self, *args: P.args, **kwds: P.kwargs) -> R:
        key = self.hash(*args, **kwds)
        cache = self._cache()
        if key not in cache:
            cache[key] = self._fn(*args, **kwds)
        return cache[key]

    def invalidate(self, *args: P.args, **kwds: P.kwargs) -> None:
        self._cache().pop(self.hash(*args, **kwds), None)
        if hasattr(self._fn, "invalidate"):
            self._fn.invalidate(*args, **kwds)

    def cache_clear(self) -> None:
        self._cache().clear()

    @staticmethod
    def hash(*args: P.args, **kwds: P.kwargs) -> tuple[Any, ...]:
        return sum(sorted(kwds.items()), args)


def set_ctx(ctx: CacheContext) -> None:
    _local.ctx = ctx


def get_thread_pool(
    max_workers: int | None = None, thread_name_prefix: str = ""
) -> ThreadPoolExecutor:
    return ThreadPoolExecutor(
        max_workers,
        thread_name_prefix,
        initializer=set_ctx,
        initargs=(_local.ctx,),
    )

## Conclusion

Implementing a per-request cache decorator in Django requires careful consideration of concurrency and thread management. By using thread-local storage and proper thread initialization, we've created a solution that works across multiple threads while maintaining request isolation.