Spaces:
Sleeping
Sleeping
File size: 2,093 Bytes
c0cb280 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | 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}")
|