scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-10 03:28:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 86

refactor(proxy): expose engine_generation/recreate_in_flight on PrismaWrapper

Browse source 
Reads the planned-recreate coordination through public PrismaWrapper accessors instead of touching private fields from PrismaClient, keeping reportPrivateUsage within the basedpyright budget.
This commit is contained in:
Devin AI 2026-08-05 17:47:19 +00:00
parent de37af0bd4
commit 3cbbf4fef6
2 changed files with 10 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/proxy/db/prisma_client.py
View file

@ -241,6 +241,14 @@ class PrismaWrapper:
self._engine_generation: int = 0
self.on_engine_replaced: Callable[[], None] | None = None
@property
def engine_generation(self) -> int:
return self._engine_generation
@property
def recreate_in_flight(self) -> bool:
return self._reconnection_lock.locked()
@staticmethod
def _read_engine(prisma_client: _PrismaClient) -> _PrismaEngine:
return prisma_client._engine

4
litellm/proxy/utils.py
View file

@ -4971,12 +4971,12 @@ class PrismaClient:
return (self.db,)
def _engine_generations(self) -> tuple[int, ...]:
return tuple(w._engine_generation for w in self._prisma_wrappers())
return tuple(w.engine_generation for w in self._prisma_wrappers())
def _is_planned_engine_recreate_error(self, e: Exception, generations_before: tuple[int, ...]) -> bool:
if not PrismaDBExceptionHandler.is_database_transport_error(e):
return False
if any(w._reconnection_lock.locked() for w in self._prisma_wrappers()):
if any(w.recreate_in_flight for w in self._prisma_wrappers()):
return True
return self._engine_generations() != generations_before