Skip to content

Context Storage API

Module: wepositive_di.context

ContextStorage

Bases: ABC

ContextStorage interface.

This interface allows for different storage backends (in-memory, Redis, etc.) to be used for storing context data, keyed by both context type and UUID.

One storage instance can hold multiple context types simultaneously. Implementations must be thread-safe and async-safe.

get_context is an async context manager that yields the context while holding a lock, ensuring safe modifications during the entire usage period.

Source code in src/wepositive_di/context.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class ContextStorage(ABC):
    """ContextStorage interface.

    This interface allows for different storage backends (in-memory, Redis, etc.)
    to be used for storing context data, keyed by both context type and UUID.

    One storage instance can hold multiple context types simultaneously.
    Implementations must be thread-safe and async-safe.

    get_context is an async context manager that yields the context while holding
    a lock, ensuring safe modifications during the entire usage period.
    """

    @abstractmethod
    def get_context(
        self, ctx_type: type[ContextTypeT], context_id: UUID
    ) -> AbstractAsyncContextManager[ContextTypeT]:
        """Get a context for the given type and context_id.

        This is an async context manager that yields the context while holding a lock.
        The lock is held until the context manager exits.

        Args:
            ctx_type: The type of context to retrieve
            context_id: The UUID identifying the context

        Yields:
            The context associated with this type and identifier

        Raises:
            KeyError: If the context does not exist
        """
        ...

    @abstractmethod
    async def store_context(
        self, ctx_type: type[ContextTypeT], context_id: UUID, context: ContextTypeT
    ) -> None:
        """Store a new context.

        This creates or replaces a context for the given type and context_id.
        Thread-safe and async-safe.

        Args:
            ctx_type: The type of context being stored
            context_id: The UUID identifying the context
            context: The context to store
        """
        ...

    @abstractmethod
    async def get_context_snapshot(
        self, ctx_type: type[ContextTypeT], context_id: UUID
    ) -> ContextTypeT:
        """Get a read-only snapshot source without taking the context lock.

        This is intended for event emission paths that must not wait behind a
        long-running mutable context lock.
        """
        ...

get_context(ctx_type, context_id) abstractmethod

Get a context for the given type and context_id.

This is an async context manager that yields the context while holding a lock. The lock is held until the context manager exits.

Parameters:

Name Type Description Default
ctx_type type[ContextTypeT]

The type of context to retrieve

 required
context_id UUID

The UUID identifying the context

 required

Yields:

Type Description
AbstractAsyncContextManager[ContextTypeT]

The context associated with this type and identifier

Raises:

Type Description
KeyError

If the context does not exist
Source code in src/wepositive_di/context.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@abstractmethod
def get_context(
    self, ctx_type: type[ContextTypeT], context_id: UUID
) -> AbstractAsyncContextManager[ContextTypeT]:
    """Get a context for the given type and context_id.

    This is an async context manager that yields the context while holding a lock.
    The lock is held until the context manager exits.

    Args:
        ctx_type: The type of context to retrieve
        context_id: The UUID identifying the context

    Yields:
        The context associated with this type and identifier

    Raises:
        KeyError: If the context does not exist
    """
    ...

store_context(ctx_type, context_id, context) abstractmethod async

Store a new context.

This creates or replaces a context for the given type and context_id. Thread-safe and async-safe.

Parameters:

Name Type Description Default
ctx_type type[ContextTypeT]

The type of context being stored

 required
context_id UUID

The UUID identifying the context

 required
context ContextTypeT

The context to store

 required
Source code in src/wepositive_di/context.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@abstractmethod
async def store_context(
    self, ctx_type: type[ContextTypeT], context_id: UUID, context: ContextTypeT
) -> None:
    """Store a new context.

    This creates or replaces a context for the given type and context_id.
    Thread-safe and async-safe.

    Args:
        ctx_type: The type of context being stored
        context_id: The UUID identifying the context
        context: The context to store
    """
    ...

get_context_snapshot(ctx_type, context_id) abstractmethod async

Get a read-only snapshot source without taking the context lock.

This is intended for event emission paths that must not wait behind a long-running mutable context lock.

Source code in src/wepositive_di/context.py 
65
66
67
68
69
70
71
72
73
74
@abstractmethod
async def get_context_snapshot(
    self, ctx_type: type[ContextTypeT], context_id: UUID
) -> ContextTypeT:
    """Get a read-only snapshot source without taking the context lock.

    This is intended for event emission paths that must not wait behind a
    long-running mutable context lock.
    """
    ...

InMemoryContextStorage

Bases: ContextStorage

Unified in-memory storage for contexts that works in both async and threaded environments.

Uses aiologic.RLock for synchronization, which works seamlessly across: - Pure async servers (FastAPI with single event loop) - Threaded servers with multiple threads - Hybrid environments (multiple threads each with their own event loop)

Contexts are stored in a two-level dict keyed first by context type, then by UUID. Fine-grained per-(type, id) locking allows concurrent access to different contexts.

Note: This implementation is single-process only. For multi-process deployments (e.g., gunicorn with multiple processes), consider using a distributed storage backend like Redis.

Source code in src/wepositive_di/context.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class InMemoryContextStorage(ContextStorage):
    """Unified in-memory storage for contexts that works in both async and threaded environments.

    Uses aiologic.RLock for synchronization, which works seamlessly across:
    - Pure async servers (FastAPI with single event loop)
    - Threaded servers with multiple threads
    - Hybrid environments (multiple threads each with their own event loop)

    Contexts are stored in a two-level dict keyed first by context type, then by UUID.
    Fine-grained per-(type, id) locking allows concurrent access to different contexts.

    Note: This implementation is single-process only. For multi-process
    deployments (e.g., gunicorn with multiple processes), consider using
    a distributed storage backend like Redis.
    """

    def __init__(self) -> None:
        self._states: dict[type[BaseModel], dict[UUID, BaseModel]] = {}
        self._locks: dict[type[BaseModel], dict[UUID, aiologic.RLock]] = {}
        self._locks_lock = aiologic.RLock()

    async def _get_lock(
        self, ctx_type: type[BaseModel], context_id: UUID
    ) -> aiologic.RLock:
        """Get or create a lock for the given (ctx_type, context_id) pair."""
        async with self._locks_lock:
            if ctx_type not in self._locks:
                self._locks[ctx_type] = {}
            if context_id not in self._locks[ctx_type]:
                self._locks[ctx_type][context_id] = aiologic.RLock()
            return self._locks[ctx_type][context_id]

    @asynccontextmanager
    async def get_context(  # pyright: ignore [reportReturnType]
        self, ctx_type: type[ContextTypeT], context_id: UUID
    ) -> AsyncGenerator[ContextTypeT]:
        lock = await self._get_lock(ctx_type, context_id)
        async with lock:
            type_store = self._states.get(ctx_type, {})
            if context_id not in type_store:
                raise KeyError(f"No {ctx_type.__name__} context known for {context_id}")
            yield type_store[context_id]  # pyright: ignore [reportReturnType]

    async def store_context(
        self,
        ctx_type: type[ContextTypeT],
        context_id: UUID,
        context: ContextTypeT,
    ) -> None:
        """Store a new context.

        Thread-safe creation/replacement of context.
        Acquires the fine-grained lock for this (ctx_type, context_id) pair.
        """
        lock = await self._get_lock(ctx_type, context_id)
        async with lock:
            if ctx_type not in self._states:
                self._states[ctx_type] = {}
            self._states[ctx_type][context_id] = context

    async def get_context_snapshot(
        self,
        ctx_type: type[ContextTypeT],
        context_id: UUID,
    ) -> ContextTypeT:
        type_store = self._states.get(ctx_type, {})
        if context_id not in type_store:
            raise KeyError(f"No {ctx_type.__name__} context known for {context_id}")
        context = type_store[context_id]
        return context.model_copy(deep=True)  # pyright: ignore [reportReturnType]

store_context(ctx_type, context_id, context) async

Store a new context.

Thread-safe creation/replacement of context. Acquires the fine-grained lock for this (ctx_type, context_id) pair.

Source code in src/wepositive_di/context.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def store_context(
    self,
    ctx_type: type[ContextTypeT],
    context_id: UUID,
    context: ContextTypeT,
) -> None:
    """Store a new context.

    Thread-safe creation/replacement of context.
    Acquires the fine-grained lock for this (ctx_type, context_id) pair.
    """
    lock = await self._get_lock(ctx_type, context_id)
    async with lock:
        if ctx_type not in self._states:
            self._states[ctx_type] = {}
        self._states[ctx_type][context_id] = context

context_storage_singleton()

Singleton provider for the context storage.

Returns the same InMemoryContextStorage instance for the lifetime of the application. One instance can hold all context types, keyed by (type, UUID).

This implementation uses aiologic.RLock which works seamlessly in: - Async servers (FastAPI): Non-blocking async synchronization - Threaded servers: Thread-safe synchronization - Hybrid environments: Multiple threads with event loops per thread

For multi-process deployments, replace with RedisContextStorage or another distributed storage implementation.

Source code in src/wepositive_di/context.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@register_provider(singleton=True)
def context_storage_singleton() -> InMemoryContextStorage:
    """Singleton provider for the context storage.

    Returns the same InMemoryContextStorage instance for the lifetime of the application.
    One instance can hold all context types, keyed by (type, UUID).

    This implementation uses aiologic.RLock which works seamlessly in:
    - Async servers (FastAPI): Non-blocking async synchronization
    - Threaded servers: Thread-safe synchronization
    - Hybrid environments: Multiple threads with event loops per thread

    For multi-process deployments, replace with RedisContextStorage or another
    distributed storage implementation.
    """
    return InMemoryContextStorage()