diff --git a/litellm/integrations/clickhouse/clickhouse_client.py b/litellm/integrations/clickhouse/clickhouse_client.py index 859a44cb803..6e071944eed 100644 --- a/litellm/integrations/clickhouse/clickhouse_client.py +++ b/litellm/integrations/clickhouse/clickhouse_client.py @@ -7,7 +7,7 @@ Uses LiteLLM's shared httpx client — no clickhouse driver dependency. import gzip import json -from typing import Any, Dict, List, Optional +from typing import Any from litellm.llms.custom_httpx.http_handler import get_async_httpx_client from litellm.types.llms.custom_http import httpxSpecialProvider @@ -20,21 +20,15 @@ class ClickHouseClient: self.url = url.rstrip("/") + "/" self.database = database self.auth_headers = {"X-ClickHouse-User": user, "X-ClickHouse-Key": password} - self.http = get_async_httpx_client( - llm_provider=httpxSpecialProvider.LoggingCallback - ) + self.http = get_async_httpx_client(llm_provider=httpxSpecialProvider.LoggingCallback) async def execute(self, sql: str) -> None: await self.http.post(self.url, content=sql.encode(), headers=self.auth_headers) - async def insert_json_each_row( - self, table: str, rows: List[Dict[str, Any]] - ) -> None: + async def insert_json_each_row(self, table: str, rows: list[dict[str, Any]]) -> None: if not rows: return - body = gzip.compress( - "\n".join(json.dumps(r, default=str) for r in rows).encode() - ) + body = gzip.compress("\n".join(json.dumps(r, default=str) for r in rows).encode()) await self.http.post( self.url, params={ @@ -48,16 +42,10 @@ class ClickHouseClient: headers={**self.auth_headers, "Content-Encoding": "gzip"}, ) - async def query( - self, sql: str, params: Optional[Dict[str, Any]] = None - ) -> List[Dict[str, Any]]: + async def query(self, sql: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: """Parameterized SELECT. Use `{name:Type}` placeholders in `sql`.""" query_params = { - f"param_{k}": ( - "[" + ",".join(f"'{x}'" for x in v) + "]" - if isinstance(v, list) - else str(v) - ) + f"param_{k}": ("[" + ",".join(f"'{x}'" for x in v) + "]" if isinstance(v, list) else str(v)) for k, v in (params or {}).items() } response = await self.http.post(