Python库 | aioreactive-0.14.0.tar.gz

<img src="logo/logo.jpg" alt="drawing" width="200"/> # aioreactive - ReactiveX for asyncio using async and await [](https://pypi.python.org/pypi/aioreactive)   [](https://codecov.io/gh/dbrattli/aioreactive) > *NEWS: Project rebooted Nov. 2020. Rebuilt using [Expression](https://github.com/dbrattli/Expression).* Aioreactive is [RxPY](https://github.com/ReactiveX/RxPY) for asyncio. It's an asynchronous and reactive Python library for asyncio using async and await. Aioreactive is built on the [Expression](https://github.com/dbrattli/Expression) functional library and, integrates naturally with the Python language. > aioreactive is the unification of RxPY and reactive programming with > asyncio using async and await. ## The design goals for aioreactive: * Python 3.9+ only. We have a hard dependency [PEP 585](https://www.python.org/dev/peps/pep-0585/), Type Hinting Generics In Standard Collections, data classes and type variables. * All operators and tools are implemented as plain old functions. * Everything is `async`. Sending values is async, subscribing to observables is async. Disposing subscriptions is async. * One scheduler to rule them all. Everything runs on the asyncio base event-loop. * No multi-threading. Only async and await with concurrency using asyncio. Threads are hard, and in many cases it doesn’t make sense to use multi-threading in Python applications. If you need to use threads you may wrap them with [`concurrent.futures`](https://docs.python.org/3/library/concurrent.futures.html#module-concurrent.futures) and compose them into the chain with `flat_map()` or similar. See [`parallel.py`](https://github.com/dbrattli/aioreactive/blob/master/examples/parallel/parallel.py) for an example. * Simple, clean and use few abstractions. Try to align with the itertools package, and reuse as much from the Python standard library as possible. * Support type hints and static type checking using [Pylance](https://devblogs.microsoft.com/python/announcing-pylance-fast-feature-rich-language-support-for-python-in-visual-studio-code/). * Implicit synchronous back-pressure &trade;. Producers of events will simply be awaited until the event can be processed by the down-stream consumers. ## AsyncObservable and AsyncObserver With aioreactive you subscribe observers to observables, and the key abstractions of aioreactive can be seen in this single line of code: ```python subscription = await observable.subscribe_async(observer) ``` The difference from RxPY can be seen with the `await` expression. Aioreactive is built around the asynchronous duals, or opposites of the AsyncIterable and AsyncIterator abstract base classes. These async classes are called AsyncObservable and AsyncObserver. AsyncObservable is a producer of events. It may be seen as the dual or opposite of AsyncIterable and provides a single setter method called `subscribe_async()` that is the dual of the `__aiter__()` getter method: ```python from abc import ABC, abstractmethod class AsyncObservable(ABC): @abstractmethod async def subscribe_async(self, observer): return NotImplemented ``` AsyncObserver is a consumer of events and is modeled after the so-called [consumer interface](http://effbot.org/zone/consumer.htm), the enhanced generator interface in [PEP-342](https://www.python.org/dev/peps/pep-0342/) and async generators in [PEP-525](https://www.python.org/dev/peps/pep-0525/). It is the dual of the AsyncIterator `__anext__()` method, and expands to three async methods `asend()`, that is the opposite of `__anext__()`, `athrow()` that is the opposite of an `raise Exception()` and `aclose()` that is the opposite of `raise StopAsyncIteration`: ```python from abc import ABC, abstractmethod class AsyncObserver(ABC): @abstractmethod async def asend(self, value): return NotImplemented @abstractmethod async def athrow(self, error): return NotImplemented @abstractmethod async def aclose(self): return NotImplemented ``` ## Subscribing to observables An observable becomes hot and starts streaming items by using the `subscribe_async()` method. The `subscribe_async()` method takes an observable and returns a disposable subscription. So the `subscribe_async()` method is used to attach a observer to the observable. ```python async def asend(value): print(value) disposable = await subscribe(source, AsyncAnonymousObserver(asend)) ``` `AsyncAnonymousObserver` is an anonymous observer that constructs an `AsyncObserver` out of plain async functions, so you don't have to implement a new named observer every time you need one. The subscription returned by `subscribe_async()` is disposable, so to unsubscribe you need to await the `dispose_async()` method on the subscription. ```python await subscription.dispose_async() ``` ## Asynchronous iteration Even more interesting, with `to_async_iterable` you can flip around from `AsyncObservable` to an `AsyncIterable` and use `async-for` to consume the stream of events. ```python obv = AsyncIteratorObserver() subscription = subscribe(source, obv) async for x in obv: print(x) ``` They effectively transform us from an async push model to an async pull model, and lets us use the awesome new language features such as `async for` and `async-with`. We do this without any queueing, as a push by the `AsyncObservable` will await the pull by the `AsyncIterator. This effectively applies so-called "back-pressure" up the subscription as the producer will await the iterator to pick up the item send. The for-loop may be wrapped with async-with to control the lifetime of the subscription: ```python import aioreactive as rx xs = rx.from_iterable([1, 2, 3]) result = [] obv = rx.AsyncIteratorObserver() async with await xs.subscribe(obv) as subscription: async for x in obv: result.append(x) assert result == [1, 2, 3] ``` ## Async streams An async stream is both an async observer and an async observable. Aioreactive lets you create streams explicitly. ```python import aioreactive as rx stream = AsyncSubject() # Alias for AsyncMultiStream sink = rx.AsyncAnonymousObserver() await stream.subscribe_async(sink) await stream.asend(42) ``` You can create streams directly from `AsyncMultiStream` or `AsyncSingleStream`. `AsyncMultiStream` supports multiple observers, and is hot in the sense that it will drop any event that is sent if there are currently no observers attached. `AsyncSingleStream` on the other hand supports a single observer, and is cold in the sense that it will await any producer until there is an observer attached. ## Operators The Rx operators in aioreactive are all plain old functions. You can apply them to an observable and compose it into a transformed, filtered, aggregated or combined observable. This transformed observable can be streamed into an observer. Observable -> Operator -> Operator -> Operator -> Observer Aioreactive contains many of the same operators as you know from RxPY. Our goal is not to implement them all, but to provide the most essential ones. * **concat** -- Concatenates two or more observables. * **choose** -- Filters and/or transforms the observable. * **choose_asnc** -- Asynchronously filters and/or transforms the observable. * **debounce** -- Throttles an observable. * **delay** -- delays the items within an observable. * **distinct_until_changed** -- an o