L'attuale .aggregate()
il metodo non è di per sé "asincrono". Ma l'iterazione del cursore lo è.
Quindi invece:
cursor = db.activities.aggregate(pipeline)
while (yield cursor.fetch_next):
doc = cursor.next_object()
print(doc)
Proprio come dicono i documenti.
