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

style(agent-tracing): apply ruff lint to clickhouse writer

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

14
litellm/integrations/clickhouse/clickhouse_writer.py
View file

@ -7,7 +7,7 @@ enqueue or are told the buffer is full (the OTLP endpoint turns that into a 429)
import asyncio
from collections import defaultdict
from typing import Any, Dict, List
from typing import Any
from litellm._logging import verbose_logger
from litellm.constants import (
@ -22,16 +22,16 @@ from litellm.integrations.clickhouse.clickhouse_client import ClickHouseClient
class ClickHouseWriter:
def __init__(self, client: ClickHouseClient):
self.client = client
self.buffers: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
self.rows_written: Dict[str, int] = defaultdict(int)
self.rows_dropped: Dict[str, int] = defaultdict(int)
self.buffers: dict[str, list[dict[str, Any]]] = defaultdict(list)
self.rows_written: dict[str, int] = defaultdict(int)
self.rows_dropped: dict[str, int] = defaultdict(int)
self._flush_lock = asyncio.Lock()
self._task: "asyncio.Task | None" = None
self._task: asyncio.Task | None = None
def is_full(self, table: str) -> bool:
return len(self.buffers[table]) >= CLICKHOUSE_MAX_BUFFERED_ROWS
def enqueue(self, table: str, rows: List[Dict[str, Any]]) -> None:
def enqueue(self, table: str, rows: list[dict[str, Any]]) -> None:
self.buffers[table].extend(rows)
def start(self) -> None:
@ -51,7 +51,7 @@ class ClickHouseWriter:
del self.buffers[table][: len(batch)]
await self._insert_with_retry(table, batch)
async def _insert_with_retry(self, table: str, batch: List[Dict[str, Any]]) -> None:
async def _insert_with_retry(self, table: str, batch: list[dict[str, Any]]) -> None:
for attempt in range(CLICKHOUSE_MAX_RETRIES):
try:
await self.client.insert_json_each_row(table, batch)