LockingProxy for thread-safe object wrapping

Wrap any Python object with LockingProxy to make all attribute access, iteration, item access, and calls thread-safe.

LockingProxy wraps an arbitrary object and intercepts every interaction with it—attribute access, method calls, iteration, item read/write, and context management—acquiring an RLock before each operation.

See the author’s article for an in-depth explanation of the design.

Quick start

from alexber.utils.thread_locals import LockingProxy

my_list = [1, 2, 3]
proxy = LockingProxy(obj=my_list)

# All operations below acquire the lock automatically
proxy.append(4)
for item in proxy:     # thread-safe iteration
    print(item)
print(proxy[0])        # thread-safe __getitem__

All arguments are keyword-only. If no lock is supplied, one is created automatically.

obj · Any · required

The object to wrap. Can be any Python object: a list, dict, callable, async generator, Pydantic model, etc.

lock · RLock

An existing RLock to use. If omitted, LockingDefaultLockMixin creates a new RLock() automatically.

Mixin architecture

LockingProxy is assembled from composable mixins. Each mixin handles one category of interaction. You can build a lighter proxy by subclassing only the mixins you need.

LockingIterableMixin

Makes __iter__ thread-safe by returning a LockingIterator that acquires the lock on each call to __next__.

from alexber.utils.thread_locals import LockingIterableMixin, LockingDefaultLockMixin, RootMixin

class SafeIterable(LockingDefaultLockMixin, LockingIterableMixin, RootMixin):
    pass

proxy = SafeIterable(obj=[10, 20, 30])
for v in proxy:
    print(v)  # each next() is lock-protected

LockingAsyncIterableMixin

Makes __aiter__ thread-safe by returning a LockingAsyncIterator that acquires the lock (async with) on each __anext__.

import asyncio
from alexber.utils.thread_locals import LockingAsyncIterableMixin, LockingDefaultLockMixin, RootMixin

class SafeAsyncIterable(LockingDefaultLockMixin, LockingAsyncIterableMixin, RootMixin):
    pass

async def run():
    proxy = SafeAsyncIterable(obj=my_async_gen())
    async for v in proxy:
        print(v)

asyncio.run(run())

LockingAccessMixin

Intercepts __getattr__ to wrap every attribute lookup:

• Regular methods are wrapped in a synchronized_method that holds the lock for the duration of the call.
• Coroutine methods are wrapped in an asynchronized_method that holds the async lock (async with) while awaiting.
• Properties / descriptors are accessed directly (no wrapping needed).

from alexber.utils.thread_locals import LockingProxy

class Counter:
    def __init__(self): self.n = 0
    def increment(self): self.n += 1
    async def async_increment(self): self.n += 1

proxy = LockingProxy(obj=Counter())
proxy.increment()          # sync — lock held during call
# await proxy.async_increment()  # async — async lock held during await

LockingCallableMixin

Intercepts __call__. If the wrapped object is a coroutine function, it is called inside async with self._lock; otherwise inside with self._lock.

from alexber.utils.thread_locals import LockingProxy

def add(a, b): return a + b

proxy = LockingProxy(obj=add)
result = proxy(1, 2)   # lock held during the call

LockingGetItemMixin

Provides thread-safe __getitem__ and __setitem__.

from alexber.utils.thread_locals import LockingProxy

d = {'key': 'value'}
proxy = LockingProxy(obj=d)
v = proxy['key']          # lock held during read
proxy['key'] = 'new'      # lock held during write

LockingSetItemMixin

Provides a thread-safe __setitem__ only (lighter alternative to LockingGetItemMixin when reads do not need protection).

SyncContextManagerMixin

Adds __enter__ / __exit__ that call self._lock.acquire() and self._lock.release(). Lets the proxy itself be used as a context manager.

with proxy:
    proxy.update_state()

AsyncContextManagerMixin

Adds __aenter__ / __aexit__ using self._lock.async_acquire() and self._lock.async_release().

async with proxy:
    await proxy.async_update_state()

LockingDefaultLockMixin

Checks whether a lock keyword argument was supplied. If not, it creates a new RLock() and injects it into kwargs before calling super().__init__. This is always included in LockingProxy.

# Explicit lock
lock = RLock()
proxy = LockingProxy(obj=my_obj, lock=lock)

# Auto-created lock (default)
proxy = LockingProxy(obj=my_obj)

Full class hierarchy

LockingProxy inherits from all the mixins above in this order:

class LockingProxy(
    LockingDefaultAndBaseLanguageModelMixin,  # auto-lock + optional LangChain support
    LockingIterableMixin,
    LockingAsyncIterableMixin,
    LockingAccessMixin,
    LockingCallableMixin,
    LockingGetItemMixin,
    LockingSetItemMixin,
    SyncContextManagerMixin,
    AsyncContextManagerMixin,
):
    ...

Python’s MRO ensures each mixin’s __init__ is called exactly once via cooperative super().__init__(**kwargs).

If you only need a subset of protections (e.g., you only iterate and never call), build a custom proxy from the individual mixins to avoid unnecessary overhead.

LockingAccessMixin uses __getattr__, which is only invoked when the normal attribute lookup fails. Internal attributes such as _obj, _lock, and _is_pedantic_obj are accessed directly without the lock.