Escape a value for use in a ClickHouse query
(value: str)
| 374 | |
| 375 | |
| 376 | def clickhouse_escape_value(value: str) -> Any: |
| 377 | """Escape a value for use in a ClickHouse query""" |
| 378 | # we don't pass the query to the full escape because it seems like `insert` |
| 379 | # handles most of it. |
| 380 | if value is None: |
| 381 | return 'NULL' |
| 382 | if isinstance(value, list): |
| 383 | return [clickhouse_escape_value(v) for v in value] |
| 384 | if isinstance(value, dict): |
| 385 | return {k: clickhouse_escape_value(v) for k, v in value.items()} |
| 386 | if isinstance(value, int): |
| 387 | return str(value) |
| 388 | return value |
| 389 | |
| 390 | |
| 391 | async def clickhouse_create(data: list[dict]) -> None: |
no outgoing calls
no test coverage detected
searching dependent graphs…