Save action event data to ClickHouse.
(data: dict)
| 93 | |
| 94 | # data['session_id'] = '06246901-8691-4ae3-848e-69aec2d0722b' |
| 95 | async def create_action_event(data: dict) -> None: |
| 96 | """Save action event data to ClickHouse.""" |
| 97 | assert 'session_id' in data, 'session_id must be provided to create action event data.' |
| 98 | assert 'agent_id' in data, 'agent_id must be provided to create action event data.' |
| 99 | project_id = await _get_project_id(data['session_id']) |
| 100 | |
| 101 | action_event = ActionEvent(**data) |
| 102 | span: Span = await action_event.to_span( |
| 103 | trace_id=str(data['session_id']), parent_span_id=str(data['agent_id']), project_id=str(project_id) |
| 104 | ) |
| 105 | await clickhouse_create_span(span) |
| 106 | logger.info(f"Created ClickHouse action event as span {span}") |
| 107 | |
| 108 | |
| 109 | async def update_action_event(data: dict) -> None: |
nothing calls this directly
no test coverage detected
searching dependent graphs…