El .aggregate()
El método no es en sí mismo "asincrónico". Pero la iteración del cursor es.
Así que en su lugar:
cursor = db.activities.aggregate(pipeline)
while (yield cursor.fetch_next):
doc = cursor.next_object()
print(doc)
Tal como dicen los documentos.
