Function clickhouse_create

app/api/agentops/exporter/processor.py:391–406

Create a record in ClickHouse

(data: list[dict])

async def clickhouse_create(data: list[dict]) -> None:
    """Create a record in ClickHouse"""
    client = await get_async_clickhouse()

    for i, row in enumerate(data):
        for key, value in row.items():
            data[i][key] = clickhouse_escape_value(value)

    if not DRY_RUN:
        await client.insert(
            table=IMPORT_TABLE_NAME,
            data=[list(row.values()) for row in data],
            column_names=list(data[0].keys()),
        )
    else:
        print(data)


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