在 Python 中实现可观察集合的推荐方法？

Question

我想在 Python 中有一些可观察的 collections/sequences 允许我监听更改事件，例如添加新项目或更新项目：

list = ObservableList(['a','b','c'])
list.addChangeListener(lambda new_value: print(new_value))
list.append('a') # => should trigger the attached change listener

data_frame = ObservableDataFrame({'x': [1,2,3], 'y':[10,20,30]})
data_frame.addChangeListener(update_dependent_table_cells) # => allows to only update dependent cells instead of a whole table

一个。我发现以下项目提供了可观察集合的实现并且看起来很有前途：

https://github.com/dimsf/Python-observable-collections

它做我想做的事：

from observablelist import ObservableList

def listHandler(event):
    if event.action == 'itemsUpdated':
        print event.action + ', old items: ' + str(event.oldItems) + ' new items: ' + str(event.newItems) + ' at index: ' + str(event.index)
    elif event.action == 'itemsAdded' or event.action == 'itemsRemoved':
        print(event.action + ', items: ' + str(event.items) + ' at index: ' + str(event.index))

myList = ObservableList()
myList.attach(listHandler)

#Do some mutation actions, just like normal lists.
myList.append(10)
myList.insert(3, 0)

然而，最后一次更改是 6 年前，我想知道是否有更多更新或 内置 Python 替代品?

乙。我还找到了 RxPy：https://github.com/ReactiveX/RxPY

import rx
list = ["Alpha", "Beta", "Gamma"]
source = rx.from_(list)
source.subscribe(
   lambda value: print(value),
   on_error = lambda e: print("Error : {0}".format(e)),
   on_completed = lambda: print("Job Done!")
)

是否有可能保持订阅打开，以便我能够在订阅订阅后将新值附加到列表？虚拟代码：

source.subscribe(..., keep_open = True)
source.append("Delta")  # <= does not work; there is no append method
source.close()

换句话说：can/should我使用 RxPy 源作为可观察集合？

C。 Python 中似乎存在许多不同的可能性来处理事件和实施观察者模式：

Event system in Python

Python Observer Pattern: Examples, Tips?

=> recommented/pythonic 在 Python 中实现可观察集合的方法是什么？我应该使用（过时的？）A. 或 B. 的改编形式（这似乎有不同的目的？）或者甚至是 C. 的另一种策略？

=> 是否有计划以某种方式标准化这种可能性并直接在 Python 中包含默认的可观察集合？

相关问题，特定于 DataFrames：

How to make tables/spreadsheets (e.g. pandas DataFrame) observable, use triggers or change events?

Answer 1

刚刚找到一个基于 RxPy 的实现。最后一次更改是从 2018 年开始的，它似乎还没有为 RxPY 3.x 做好准备。

https://github.com/shyam-s00/ObservableCollections

https://github.com/shyam-s00/ObservableCollections/issues/1

from reactive.ObservableList import ObservableList

ol = ObservableList([1, 2, 3, 4])
ol.when_collection_changes() \
    .map(lambda x: x.Items) \
    .subscribe(print, print)

ol.append(5)

提供

ObservableList
ObservableDict
ObservableSet

另见 https://github.com/ReactiveX/RxPY/issues/553

Answer 2

我从未使用过 RxPy，但它似乎是一种非常接近 js/ts 的 rx 模式的实现。

首先，你想要一个可观察的对象，你们都用它来将数据推送到它和观察者中。那是一个 subject，可能是一个行为主题或重播主题。创建主题，然后使用 on_next() 运算符将新值推送到其中。

对于你的第二个问题，你似乎想将多个可观察值“组合”成一个可观察值。有多种方法可以做到这一点，但最有可能的是，您正在寻找的是 CombineLatest 或 Concat。在 operators.

抢劫

如果我以你的第二个例子为例，代码将如下所示：

from rx.subject.subject import Subject

list = ["Alpha", "Beta", "Gamma"]
# assuming that you want each item to be emitted one after the other
subject = Subject()
subject.subscribe(
    lambda value: print(value),
    on_error = lambda e: print("Error : {0}".format(e)),
    on_completed = lambda: print("Job Done!")
)
subject.on_next('Alpha')
subject.on_next('Beta')
subject.on_next('Gamma')
subject.on_next('Delta')

如果您使用 BehaviourSubject，您将能够提供一个初始值，当一个新的观察者订阅时，它会收到最后发出的值。如果您使用 ReplaySubject，您可以提供值，然后订阅，观察者将收到主题到目前为止发出的所有值。

在 Python 中实现可观察集合的推荐方法？

Recommented way to implement observable collections in Python?

python

collections

sequence

observable

observer-pattern