Function clickhouse_update_span

app/api/agentops/exporter/processor.py:451–477

Update a record in ClickHouse by deleting and re-inserting it.

(span_id: str, update_data: dict)

async def clickhouse_update_span(span_id: str, update_data: dict) -> None:
    """Update a record in ClickHouse by deleting and re-inserting it."""
    # reasons for going this route at the moment:
    # - serialization is a bitch and there is no tooling readily available to help with it
    # - latency for propagation of updates is apparently comparable to deletion

    def merge_dicts_recursive(old, new):
        result = old.copy()
        for key, value in new.items():
            if key in result and isinstance(result[key], dict) and isinstance(value, dict):
                result[key] = merge_dicts_recursive(result[key], value)
            else:
                result[key] = value
        return result

    existing_span: dict = await clickhouse_get_span_raw(span_id)
    if existing_span:
        await clickhouse_delete_span(span_id)  # yolo
    else:
        existing_span = {}

    merged_data = merge_dicts_recursive(existing_span, update_data)
    await clickhouse_create(
        [
            merged_data,
        ]
    )


async def write_trace_with_timeout(trace: Trace) -> None: