RuJect/snakia
1
0
Fork 0
Code Issues 5 Pull requests Projects Releases 5 Packages Wiki Activity Actions

feat: more safety in BaseBindable, AsyncBindable, Bindable

Browse source
This commit is contained in:
rus07tam 2025-11-23 10:47:37 +00:00
parent 7f83370873
commit c7cd08b7e0
3 changed files with 47 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

29
src/snakia/core/rx/async_bindable.py
View file

@ -1,4 +1,12 @@
from typing import Any, Awaitable, Callable, Generic, Literal, TypeVar, overload
from typing import (
Any,
Awaitable,
Callable,
Generic,
Literal,
TypeVar,
overload,
)
from .base_bindable import BaseBindable, BindableSubscriber, ValueChanged
@ -23,6 +31,9 @@ class AsyncBindable(BaseBindable[T], Generic[T]):
async def set(self, value: T) -> None:
"""Set the value."""
if not (self.has_default_value or self.has_value):
self.set_silent(value)
return
e = ValueChanged(self.value, value)
self.set_silent(value)
for subscriber in self.__subscribers:
@ -55,19 +66,25 @@ class AsyncBindable(BaseBindable[T], Generic[T]):
if run_now:
async def _run() -> None:
await subscriber(ValueChanged(self.__default_value, self.value))
await subscriber(
ValueChanged(self.__default_value, self.value)
)
return _run()
return None
def unsubscribe(self, subscriber: BindableSubscriber[T, Awaitable[Any]]) -> None:
def unsubscribe(
self, subscriber: BindableSubscriber[T, Awaitable[Any]]
) -> None:
"""Unsubscribe from an value."""
self.__subscribers.remove(subscriber)
@overload
def on(
self, run_now: Literal[True]
) -> Callable[[BindableSubscriber[T, Awaitable[Any]]], Awaitable[None]]: ...
) -> Callable[
[BindableSubscriber[T, Awaitable[Any]]], Awaitable[None]
]: ...
@overload
def on(
@ -87,7 +104,9 @@ class AsyncBindable(BaseBindable[T], Generic[T]):
if run_now:
async def _run() -> None:
await subscriber(ValueChanged(self.__default_value, self.value))
await subscriber(
ValueChanged(self.__default_value, self.value)
)
return _run()
return None