motor -> pymongo

Author: HardMax71

backend/app/core/database_context.py

@@ -167,8 +167,8 @@ async def session(self) -> AsyncIterator[DBSession]:
             async with connection.session() as session:
                 await collection.insert_one(doc, session=session)
         """
-        async with await self.client.start_session() as session:
-            async with session.start_transaction():
+        async with self.client.start_session() as session:
+            async with await session.start_transaction():
                 yield session
 
 

backend/app/db/repositories/admin/admin_events_repository.py

@@ -120,7 +120,8 @@ async def get_event_stats(self, hours: int = 24) -> EventStatistics:
 
         # Get overview statistics
         overview_pipeline = EventStatsAggregation.build_overview_pipeline(start_time)
-        overview_result = await self.events_collection.aggregate(overview_pipeline).to_list(1)
+        overview_cursor = await self.events_collection.aggregate(overview_pipeline)
+        overview_result = await overview_cursor.to_list(1)
 
         stats = (
             overview_result[0]
@@ -140,19 +141,20 @@ async def get_event_stats(self, hours: int = 24) -> EventStatistics:
 
         # Get event types with counts
         type_pipeline = EventStatsAggregation.build_event_types_pipeline(start_time)
-        top_types = await self.events_collection.aggregate(type_pipeline).to_list(10)
+        type_cursor = await self.events_collection.aggregate(type_pipeline)
+        top_types = await type_cursor.to_list(10)
         events_by_type = {t["_id"]: t["count"] for t in top_types}
 
         # Get events by hour
         hourly_pipeline = EventStatsAggregation.build_hourly_events_pipeline(start_time)
-        hourly_cursor = self.events_collection.aggregate(hourly_pipeline)
+        hourly_cursor = await self.events_collection.aggregate(hourly_pipeline)
         events_by_hour: list[HourlyEventCount | dict[str, Any]] = [
             HourlyEventCount(hour=doc["_id"], count=doc["count"]) async for doc in hourly_cursor
         ]
 
         # Get top users
         user_pipeline = EventStatsAggregation.build_top_users_pipeline(start_time)
-        top_users_cursor = self.events_collection.aggregate(user_pipeline)
+        top_users_cursor = await self.events_collection.aggregate(user_pipeline)
         top_users = [
             UserEventCount(user_id=doc["_id"], event_count=doc["count"])
             async for doc in top_users_cursor
@@ -175,7 +177,8 @@ async def get_event_stats(self, hours: int = 24) -> EventStatistics:
             {"$group": {"_id": None, "avg_duration": {"$avg": "$resource_usage.execution_time_wall_seconds"}}},
         ]
 
-        exec_result = await executions_collection.aggregate(exec_pipeline).to_list(1)
+        exec_cursor = await executions_collection.aggregate(exec_pipeline)
+        exec_result = await exec_cursor.to_list(1)
         avg_processing_time = (
             exec_result[0]["avg_duration"] if exec_result and exec_result[0].get("avg_duration") else 0
         )

backend/app/db/repositories/dlq_repository.py

@@ -35,7 +35,7 @@ async def get_dlq_stats(self) -> DLQStatistics:
         ]
 
         status_results = []
-        async for doc in self.dlq_collection.aggregate(status_pipeline):
+        async for doc in await self.dlq_collection.aggregate(status_pipeline):
             status_results.append(doc)
 
         # Convert status results to dict
@@ -58,7 +58,7 @@ async def get_dlq_stats(self) -> DLQStatistics:
         ]
 
         by_topic: List[TopicStatistic] = []
-        async for doc in self.dlq_collection.aggregate(topic_pipeline):
+        async for doc in await self.dlq_collection.aggregate(topic_pipeline):
             by_topic.append(
                 TopicStatistic(topic=doc["_id"], count=doc["count"], avg_retry_count=round(doc["avg_retry_count"], 2))
             )
@@ -71,7 +71,7 @@ async def get_dlq_stats(self) -> DLQStatistics:
         ]
 
         by_event_type: List[EventTypeStatistic] = []
-        async for doc in self.dlq_collection.aggregate(event_type_pipeline):
+        async for doc in await self.dlq_collection.aggregate(event_type_pipeline):
             if doc["_id"]:  # Skip null event types
                 by_event_type.append(EventTypeStatistic(event_type=doc["_id"], count=doc["count"]))
 
@@ -94,7 +94,8 @@ async def get_dlq_stats(self) -> DLQStatistics:
             },
         ]
 
-        age_result = await self.dlq_collection.aggregate(age_pipeline).to_list(1)
+        age_cursor = await self.dlq_collection.aggregate(age_pipeline)
+        age_result = await age_cursor.to_list(1)
         age_stats_data = age_result[0] if age_result else {}
         age_stats = AgeStatistics(
             min_age_seconds=age_stats_data.get("min_age", 0.0),
@@ -148,7 +149,7 @@ async def get_topics_summary(self) -> list[DLQTopicSummary]:
         ]
 
         topics = []
-        async for result in self.dlq_collection.aggregate(pipeline):
+        async for result in await self.dlq_collection.aggregate(pipeline):
             status_counts: dict[str, int] = {}
             for status in result["statuses"]:
                 status_counts[status] = status_counts.get(status, 0) + 1

backend/app/db/repositories/event_repository.py

@@ -234,7 +234,8 @@ async def get_event_statistics(
             ]
         )
 
-        result = await self._collection.aggregate(pipeline).to_list(length=1)
+        cursor = await self._collection.aggregate(pipeline)
+        result = await cursor.to_list(length=1)
 
         if result:
             stats = result[0]
@@ -296,7 +297,8 @@ async def get_event_statistics_filtered(
             ]
         )
 
-        result = await self._collection.aggregate(pipeline).to_list(length=1)
+        cursor = await self._collection.aggregate(pipeline)
+        result = await cursor.to_list(length=1)
         if result:
             stats = result[0]
             return EventStatistics(
@@ -321,7 +323,9 @@ async def stream_events(
         if filters:
             pipeline.append({"$match": filters})
 
-        async with self._collection.watch(pipeline, start_after=start_after, full_document="updateLookup") as stream:
+        async with await self._collection.watch(
+            pipeline, start_after=start_after, full_document="updateLookup"
+        ) as stream:
             async for change in stream:
                 if change["operationType"] in ["insert", "update", "replace"]:
                     yield change["fullDocument"]
@@ -438,7 +442,7 @@ async def aggregate_events(self, pipeline: list[dict[str, object]], limit: int =
         pipeline.append({"$limit": limit})
 
         results = []
-        async for doc in self._collection.aggregate(pipeline):
+        async for doc in await self._collection.aggregate(pipeline):
             if "_id" in doc and isinstance(doc["_id"], dict):
                 doc["_id"] = str(doc["_id"])
             results.append(doc)
@@ -451,7 +455,7 @@ async def list_event_types(self, match: Mapping[str, object] = MappingProxyType(
             pipeline.append({"$match": dict(match)})
         pipeline.extend([{"$group": {"_id": f"${EventFields.EVENT_TYPE}"}}, {"$sort": {"_id": 1}}])
         event_types: list[str] = []
-        async for doc in self._collection.aggregate(pipeline):
+        async for doc in await self._collection.aggregate(pipeline):
             event_types.append(doc["_id"])
         return event_types
 

backend/app/db/repositories/execution_repository.py

@@ -130,9 +130,10 @@ async def get_executions(
         executions: list[DomainExecution] = []
         async for doc in cursor:
             sv = doc.get("status")
+            resource_usage_data = doc.get("resource_usage")
             executions.append(
                 DomainExecution(
-                    execution_id=doc.get("execution_id"),
+                    execution_id=str(doc.get("execution_id", "")),
                     script=doc.get("script", ""),
                     status=ExecutionStatus(str(sv)),
                     stdout=doc.get("stdout"),
@@ -142,8 +143,8 @@ async def get_executions(
                     created_at=doc.get("created_at", datetime.now(timezone.utc)),
                     updated_at=doc.get("updated_at", datetime.now(timezone.utc)),
                     resource_usage=(
-                        ResourceUsageDomain.from_dict(doc.get("resource_usage"))
-                        if doc.get("resource_usage") is not None
+                        ResourceUsageDomain.from_dict(dict(resource_usage_data))
+                        if resource_usage_data is not None
                         else None
                     ),
                     user_id=doc.get("user_id"),

backend/app/db/repositories/notification_repository.py

@@ -25,7 +25,8 @@ def __init__(self, database: Database):
 
     async def create_indexes(self) -> None:
         # Create indexes if only _id exists
-        notif_indexes = await self.notifications_collection.list_indexes().to_list(None)
+        notif_cursor = await self.notifications_collection.list_indexes()
+        notif_indexes = await notif_cursor.to_list(None)
         if len(notif_indexes) <= 1:
             await self.notifications_collection.create_indexes(
                 [
@@ -38,7 +39,8 @@ async def create_indexes(self) -> None:
                 ]
             )
 
-        subs_indexes = await self.subscriptions_collection.list_indexes().to_list(None)
+        subs_cursor = await self.subscriptions_collection.list_indexes()
+        subs_indexes = await subs_cursor.to_list(None)
         if len(subs_indexes) <= 1:
             await self.subscriptions_collection.create_indexes(
                 [

backend/app/db/repositories/saga_repository.py

@@ -93,7 +93,7 @@ async def count_sagas_by_state(self) -> dict[str, int]:
         pipeline = [{"$group": {"_id": "$state", "count": {"$sum": 1}}}]
 
         result = {}
-        async for doc in self.sagas.aggregate(pipeline):
+        async for doc in await self.sagas.aggregate(pipeline):
             result[doc["_id"]] = doc["count"]
 
         return result
@@ -123,7 +123,7 @@ async def get_saga_statistics(self, saga_filter: SagaFilter | None = None) -> di
         state_pipeline = [{"$match": query}, {"$group": {"_id": "$state", "count": {"$sum": 1}}}]
 
         states = {}
-        async for doc in self.sagas.aggregate(state_pipeline):
+        async for doc in await self.sagas.aggregate(state_pipeline):
             states[doc["_id"]] = doc["count"]
 
         # Average duration for completed sagas
@@ -134,7 +134,7 @@ async def get_saga_statistics(self, saga_filter: SagaFilter | None = None) -> di
         ]
 
         avg_duration = 0.0
-        async for doc in self.sagas.aggregate(duration_pipeline):
+        async for doc in await self.sagas.aggregate(duration_pipeline):
             # Convert milliseconds to seconds
             avg_duration = doc["avg_duration"] / 1000.0 if doc["avg_duration"] else 0.0
 

backend/app/dlq/manager.py

@@ -334,7 +334,7 @@ async def _update_queue_metrics(self) -> None:
             {"$group": {"_id": f"${DLQFields.ORIGINAL_TOPIC}", "count": {"$sum": 1}}},
         ]
 
-        async for result in self.dlq_collection.aggregate(pipeline):
+        async for result in await self.dlq_collection.aggregate(pipeline):
             # Note: OpenTelemetry doesn't have direct gauge set, using delta tracking
             self.metrics.update_dlq_queue_size(result["_id"], result["count"])
 

backend/app/events/event_store.py

@@ -68,7 +68,8 @@ async def initialize(self) -> None:
             ),
         ]
 
-        existing = await self.collection.list_indexes().to_list(None)
+        indexes_cursor = await self.collection.list_indexes()
+        existing = await indexes_cursor.to_list(None)
         if len(existing) <= 1:
             await self.collection.create_indexes(event_indexes)
         logger.info(f"Created {len(event_indexes)} indexes for events collection")
@@ -300,7 +301,7 @@ async def get_event_stats(
             ]
         )
 
-        cursor = self.collection.aggregate(pipeline)
+        cursor = await self.collection.aggregate(pipeline)
         stats: Dict[str, Any] = {"total_events": 0, "event_types": {}, "start_time": start_time, "end_time": end_time}
         async for r in cursor:
             et = r["_id"]

backend/tests/conftest.py

@@ -11,7 +11,7 @@
 from dishka import AsyncContainer
 from dotenv import load_dotenv
 from httpx import ASGITransport
-from pymongo.asynchronous.database import AsyncDatabase as AsyncIOMotorDatabase
+from app.core.database_context import Database
 import redis.asyncio as redis
 
 # Load test environment variables BEFORE any app imports
@@ -175,8 +175,8 @@ async def scope(app_container: AsyncContainer):  # type: ignore[valid-type]
 
 
 @pytest_asyncio.fixture(scope="function")
-async def db(scope) -> AsyncGenerator[AsyncIOMotorDatabase, None]:  # type: ignore[valid-type]
-    database: AsyncIOMotorDatabase = await scope.get(AsyncIOMotorDatabase)
+async def db(scope) -> AsyncGenerator[Database, None]:  # type: ignore[valid-type]
+    database: Database = await scope.get(Database)
     yield database