codestare.async_utils.descriptor module

class codestare.async_utils.descriptor.fget_type(*args, **kwargs)

Bases: Protocol[T_contra]

__call__() → T_contra

fget_type callables need to have this signature

class codestare.async_utils.descriptor.fset_type(*args, **kwargs)

Bases: Protocol[T_co]

__call__(value: T_co) → None

fset_type callables need to have this signature

class codestare.async_utils.descriptor.accessor(*, funcs: None = ..., condition: asyncio.Condition | None = None, name: str = None)

class codestare.async_utils.descriptor.accessor(*, condition: asyncio.Condition | None = None, name: str = None)

class codestare.async_utils.descriptor.accessor(*, funcs: Tuple[fget_type[T], fset_type[T]] = ..., condition: asyncio.Condition | None = None, name: str = None)

Bases: Generic[T]

An accessor provides easy shared access to a resource and

Example

In the simplest case, an accessor synchronizes reads and writes out of the box

>>> import asyncio
>>> from codestare.async_utils import accessor
>>> foo = accessor()
>>> async def wait_for_write(accessor_):
...     print(await accessor_.get())
...
>>> background = asyncio.create_task(wait_for_write(foo))
>>> await foo.set("Bar")
Bar

It’s possible to use custom getters / setter e.g. create an accessor to the value managed by a normal property if one needs shared access as well

>>> class Thing:
...     def __init__(self):
...         self._value = None
...     @property
...     def value(self):
...         return self._value
...     @value._setter
...     def value(self, val):
...         if not val:
...             raise ValueError(f"Illegal value {val}")
...         self._value = val
...
>>> thing = Thing()
>>> thing.value = 3
>>> thing.value
3
>>> thing.value = 0
ValueError: Illegal value 0
>>> class BetterThing(Thing):
...     def __init__(self):
...         super().__init__()
...         self.value_accessor = accessor(funcs=(
...             type(self).value.fget.__get__(self),
...             type(self).value.fset.__get__(self)
...         ))
...
>>> better_thing = BetterThing()
>>> background = asyncio.create_task(wait_for_write(better_thing.value_accessor))
>>> await better_thing.value_accessor.set(3)
3
>>> better_thing.value
3
>>> await better_thing.value_accessor.set(0)
ValueError: Illegal value 0

See also

condition_property – decorator to create accessor properties more easily

__init__(*, funcs: None = None, condition: asyncio.Condition | None = None, name: str = None)

__init__(*, condition: asyncio.Condition | None = None, name: str = None)

__init__(*, funcs: Tuple[fget_type[T], fset_type[T]] = None, condition: asyncio.Condition | None = None, name: str = None)

Parameters:

• funcs – getter and setter for some value – optional, if not passed get / set a private field of the object

• condition – condition to synchronize access to the value – optional, if not passed a new condition is created

fset: fset_type[T]

Setter, either passed via funcs argument, or a setter of an internal value if no funcs where passed

fget: fget_type[T]

Getter, either passed via funcs argument, or a getter of an internal value if no funcs where passed

condition: Condition

Used to synchronized access, either passed via condition argument, or a new condition created specifically for this accessor

name

For debug purposes

has_waiting_read

Use this awaitable if you want to wait for read access

property value: T | None

Simple access to the value produced by fget without async locks i.e. not safe if you did not acquire the lock of condition

async set(value: T, wait_for_read=False) → None

Sets the value (using fset) and notifies every coroutine waiting on the condition (e.g. get()

Parameters:

• value – new value passed to fset

• wait_for_read – if True, will set the value only after futures are waiting by using get. Use this to invert the semantics – a write waiting for a read, instead of a read waiting for a write)

async get(*, predicate: Callable[[T], bool] | None = None, wait_for_write: bool | None = None) → T

Shared access to value produced by fget

Parameters:

• predicate – waits for the predicate result to be truthy, then returns the result of fget. The default predicate (used when predicate=None) returns [False, True, True, ...], so get() blocks once, until it is notified from a set() and then does not block again. Passing predicate=(lambda: True) will make get() not block at all.

• wait_for_write – if set to True, and a predicate is passed, the predicate will only be applied once the default predicate (see above) also returns True i.e. you get the next value that matches the predicate, even if the current value also matches. You can set this value to False, to ignore the default predicate behaviour (which is the same as passing predicate=(lambda: True) and using the default for this value. – optional

Returns:

value produced by fget

Raises:

ValueError – if predicate is not a callable

See also

asyncio.Condition.wait_for() – used to wait for internal condition

class codestare.async_utils.descriptor.condition_property(fget: Callable[[Any], T] | None = None, fset: Callable[[Any, T], None] | None = None, fdel: Callable[[Any], None] | None = None, doc: str | None = None)

Bases: cached_property, Generic[T]

This is a decorator to create a cached accessor to handle access to some data via a asyncio.Condition.

You can use it like the normal @property decorator, but the result of the lookup (__get__ of the descriptor) will be an accessor with coroutine attributes to handle safely setting and getting the value (from the objects methods passed via setter and getter, like in normal properties) by means of a condition.

See also

accessor – how to access the value

getter(fget: Callable[[Any], T]) → condition_property[T]

This is a cached property but uses the same interface as a normal property. See example of property documentation on how to use.

setter(fset: Callable[[Any, T], None]) → condition_property[T]

This is a cached property but uses the same interface as a normal property. See example of property documentation on how to use.

deleter(fdel) → condition_property[T]

This is a cached property but uses the same interface as a normal property. See example of property documentation on how to use.