Aktuální .aggregate()
metoda sama o sobě není "asynchronní". Ale iterace kurzoru je.
Takže místo toho:
cursor = db.activities.aggregate(pipeline)
while (yield cursor.fetch_next):
doc = cursor.next_object()
print(doc)
Přesně jak říkají dokumenty.