import asyncio from datetime import datetime import logging from analytics.models import WorkflowExecutionLog from core.database import get_db_session logger = logging.getLogger(__name__) class AsyncAnalyticsCollector: _instance = None @classmethod def get_instance(cls): if not cls._instance: cls._instance = cls() return cls._instance async def log_step(self, execution_id, workflow_id, step_id, step_type, start_time, end_time, status, error=None, trigger_data=None, results=None): """Non-blocking log submission""" try: duration = (end_time - start_time).total_seconds() * 1000 log_entry = { "execution_id": execution_id, "workflow_id": workflow_id, "step_id": step_id, "step_type": str(step_type), "start_time": start_time, "end_time": end_time, "duration_ms": duration, "status": status, "error_code": str(error) if error else None, "trigger_data": trigger_data, "results": results } # Spawn fire-and-forget task # Using asyncio.create_task to ensure it runs on the event loop without blocking asyncio.create_task(self._persist_log(log_entry)) except Exception as e: logger.error(f"Failed to queue analytics log: {e}") async def _persist_log(self, data): """Persist to DB in separate thread""" try: # Run blocking DB operation in a separate thread await asyncio.to_thread(self._sync_write, data) except Exception as e: logger.error(f"Failed to write analytics log: {e}") def _sync_write(self, data): try: with get_db_session() as db: log = WorkflowExecutionLog(**data) db.add(log) db.commit() except Exception as e: logger.error(f"DB Write Error in Analytics: {e}")