scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

style(agent-tracing): apply ruff lint + format to clickhouse client

Browse source
This commit is contained in:
Ishaan Jaff 2026-09-29 22:51:04 -07:00
parent 475b25a85b
commit a8f947c5a5
No known key found for this signature in database
1 changed files with 6 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
litellm/integrations/clickhouse/clickhouse_client.py
View file

@ -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(