scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-25 01:02:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 80

refactor(jwt): route existing-key lookup through VerificationTokenRepository

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yuneng 2026-09-22 02:12:16 +00:00
parent 18662b3c13
commit acb8466899
2 changed files with 21 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
litellm/proxy/auth/user_api_key_auth.py
View file

@ -929,21 +929,7 @@ class _PendingAutoRegister(NamedTuple):
async def _latest_active_key_hash_for_user(prisma_client: PrismaClient, user_id: str) -> str | None:
row: Final = await VerificationTokenRepository(prisma_client).table.find_first(
where={ # mutable-ok: the prisma where clause contract is a plain dict
"user_id": user_id,
"AND": [ # mutable-ok: prisma filter literal
{"OR": [{"blocked": False}, {"blocked": None}]}, # mutable-ok: prisma filter literal
{ # mutable-ok: prisma filter literal
"OR": [ # mutable-ok: prisma filter literal
{"expires": None}, # mutable-ok: prisma filter literal
{"expires": {"gt": datetime.now(timezone.utc)}}, # mutable-ok: prisma filter literal
]
},
],
},
order={"created_at": "desc"}, # mutable-ok: prisma order literal
)
row: Final = await VerificationTokenRepository(prisma_client).find_latest_active_row_by_user_id(user_id)
return None if row is None else row.token

21
litellm/repositories/verification_token_repository.py
View file

@ -4,7 +4,7 @@ VerificationToken repository for database operations on LiteLLM_VerificationToke
import json
from collections.abc import Mapping, Sequence
from datetime import datetime
from datetime import datetime, timezone
from types import TracebackType
from typing import TYPE_CHECKING, Final, Protocol
@ -123,6 +123,25 @@ class VerificationTokenRepository(BaseRepository[LiteLLM_VerificationToken]):
records: Final[Sequence[PrismaVerificationToken]] = await self.table.find_many(where={"user_id": user_id})
return self._to_model_list(records)
async def find_latest_active_row_by_user_id(self, user_id: str) -> "PrismaVerificationToken | None":
"""Find the most recently created non-blocked, non-expired token row for a user."""
row: Final = await self.table.find_first(
where={ # mutable-ok: the prisma where clause contract is a plain dict
"user_id": user_id,
"AND": [ # mutable-ok: prisma filter literal
{"OR": [{"blocked": False}, {"blocked": None}]}, # mutable-ok: prisma filter literal
{ # mutable-ok: prisma filter literal
"OR": [ # mutable-ok: prisma filter literal
{"expires": None}, # mutable-ok: prisma filter literal
{"expires": {"gt": datetime.now(timezone.utc)}}, # mutable-ok: prisma filter literal
]
},
],
},
order={"created_at": "desc"}, # mutable-ok: prisma order literal
)
return row
async def find_by_team_id(self, team_id: str) -> list[LiteLLM_VerificationToken]:
"""Find all tokens belonging to a team."""
records: Final[Sequence[PrismaVerificationToken]] = await self.table.find_many(where={"team_id": team_id})