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

Question

我希望在 Python 中有一些可观察的集合/序列，让我能够监听更改事件，例如添加新项目或更新项目：

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 替代品？

B.我还找到了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()

换句话说：我可以/应该使用 RxPy 源作为可观察的集合吗？

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

Python 中的事件系统

Python 观察者模式：示例、技巧？

在 python 中实现观察者模式的替代方法

Python3中使用装饰器实现观察者模式

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

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

特定于 DataFrame 的相关问题：

如何使表格/电子表格（例如pandas DataFrame）可观察，使用触发器或更改事件？

Answer 1

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

首先，您需要一个可观察对象，您可以使用它来将数据推送到它和观察者中。这是一个主题，可能是一个行为主题或重放主题。创建主题，然后使用 on_next() 运算符将新值推送到其中。

对于第二个问题，您似乎想将多个可观察量“组合”为一个可观察量。有多种方法可以做到这一点，但最有可能的是，您正在寻找的是CombineLatest 或Concat。掠夺经营者。

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

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，您可以提供值，然后订阅，观察者将收到主题到目前为止发出的所有值。