如何使用Rx3.1.0v和pymongo接收反应性mongo数据

Question

我试图使用Rx3.1.0获取可观测的芒果数据，我试图实现"Observable.from_“--出现了一个错误，即"from_”不是可观察的方法

from pymongo import MongoClient
from rx import Observable

def collect_data(dbname, collection):
clt = MongoClient('localhost', 27017)
db = clt[dbname]
col = db[collection]

colData= col.find().limit(10)
return Observable.from_(colData)
collect_data('test', 'money').subscribe(print(data))

出现了一个错误，即"from_“未被视为可观察的方法。

我试着用rx.of

from pymongo import MongoClient
import rx

def collect_data(dbname, collection):
clt = MongoClient('localhost', 27017)
db = clt[dbname]
col = db[collection]

colData= col.find().limit(10)
return rx.of(colData)
collect_data('test', 'money').subscribe(lambda data: print('{0}'.format(data)))

我得到了

<pymongo.cursor.Cursor object at 0x000001BF824FE8C8>

Answer 1

而不是

Observable.from_(colData)

我应该用RxPY 3.1.0的新语句替换

rx.from_(colData).pipe( ops.filter(.)，ops.distinct(.)，ops.take(.)，ops.map(.)，

Answer 2

find()返回一个可迭代的游标；因此，在与for循环或类似的猜测中使用它。例如：

for item in colData
    ...

或

list(colData)