Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/alex-ber/AlexBerUtils/llms.txt

Use this file to discover all available pages before exploring further.

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.

Build docs developers (and LLMs) love

Get started for freeTalk to us