codeflash-admin/codeflash-internal
1
0
Fork 0
mirror of https://github.com/codeflash-ai/codeflash-internal.git synced 2026-05-04 18:25:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: resolve mypy type errors in llm.py

Browse source 
- Replace Lock._loop private attribute access with explicit loop tracking
- Add explicit type annotation for clients_to_close list in shutdown()

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
claude[bot] 2026-04-06 16:24:14 +00:00
parent cada44b526
commit 7457b89631
1 changed files with 4 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
django/aiservice/aiservice/llm.py
View file

@ -76,6 +76,7 @@ class LLMClient:
self.client_loop: asyncio.AbstractEventLoop | None = None self.client_loop: asyncio.AbstractEventLoop | None = None
self.background_tasks: set[asyncio.Task[Any]] = set() self.background_tasks: set[asyncio.Task[Any]] = set()
self._client_lock: asyncio.Lock | None = None self._client_lock: asyncio.Lock | None = None
self._client_lock_loop: asyncio.AbstractEventLoop | None = None
async def call( async def call(
self, self,
@ -96,8 +97,9 @@ class LLMClient:
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
# Create lock for this event loop if it doesn't exist # Create lock for this event loop if it doesn't exist
if self._client_lock is None or self._client_lock._loop is not loop: if self._client_lock is None or self._client_lock_loop is not loop:
self._client_lock = asyncio.Lock() self._client_lock = asyncio.Lock()
self._client_lock_loop = loop
if loop is not self.client_loop: if loop is not self.client_loop:
async with self._client_lock: async with self._client_lock:
@ -243,7 +245,7 @@ class LLMClient:
This method ensures all clients are explicitly closed during graceful shutdown, This method ensures all clients are explicitly closed during graceful shutdown,
before the event loop is torn down. before the event loop is torn down.
""" """
clients_to_close = [] clients_to_close: list[tuple[str, AsyncAzureOpenAI | AsyncAnthropicBedrock]] = []
if self.openai_client is not None: if self.openai_client is not None:
clients_to_close.append(("OpenAI", self.openai_client)) clients_to_close.append(("OpenAI", self.openai_client))