From fa6a94cef13383af0ed38689d2af80161f95569d Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 17 Jul 2026 20:56:41 +0000 Subject: [PATCH] refactor(auth_v2): modernize resolvers.py to PEP 604/585 typing --- litellm/proxy/auth_v2/resolvers.py | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/litellm/proxy/auth_v2/resolvers.py b/litellm/proxy/auth_v2/resolvers.py index 18e34e90047..cef6dde408f 100644 --- a/litellm/proxy/auth_v2/resolvers.py +++ b/litellm/proxy/auth_v2/resolvers.py @@ -1,7 +1,7 @@ from __future__ import annotations import uuid -from typing import TYPE_CHECKING, List, Optional, Protocol, cast +from typing import TYPE_CHECKING, Protocol, cast from scim2_models import Group as ScimGroup from scim2_models import User as ScimUser @@ -62,13 +62,13 @@ class Resolver(Protocol): class ProvisioningStore(Protocol): async def upsert_user(self, user: ScimUser) -> ScimUser: ... - async def get_user(self, resource_id: str) -> Optional[ScimUser]: ... + async def get_user(self, resource_id: str) -> ScimUser | None: ... async def deactivate_user(self, resource_id: str) -> None: ... - async def list_users(self, filter_expr: Optional[str]) -> List[ScimUser]: ... + async def list_users(self, filter_expr: str | None) -> list[ScimUser]: ... async def upsert_group(self, group: ScimGroup) -> ScimGroup: ... - async def get_group(self, resource_id: str) -> Optional[ScimGroup]: ... + async def get_group(self, resource_id: str) -> ScimGroup | None: ... async def delete_group(self, resource_id: str) -> None: ... - async def list_groups(self, filter_expr: Optional[str]) -> List[ScimGroup]: ... + async def list_groups(self, filter_expr: str | None) -> list[ScimGroup]: ... class DbResolver(Resolver, ProvisioningStore): @@ -104,7 +104,7 @@ class DbResolver(Resolver, ProvisioningStore): raise errors.account_disabled() return self._principal_from_key(credential, key, await self._key_role(key)) - async def _key_role(self, key: UserAPIKeyAuth) -> Optional[Role]: + async def _key_role(self, key: UserAPIKeyAuth) -> Role | None: """Platform role for an API key. ``get_key_object`` does not join the owning user's role onto the token, so @@ -154,8 +154,8 @@ class DbResolver(Resolver, ProvisioningStore): credential_ref=credential.credential_ref, ) - def _principal_from_key(self, credential: Credential, key: UserAPIKeyAuth, role: Optional[Role]) -> Principal: - teams: List[TeamIdentity] = [] + def _principal_from_key(self, credential: Credential, key: UserAPIKeyAuth, role: Role | None) -> Principal: + teams: list[TeamIdentity] = [] if key.team_id is not None: membership = team_role(key.team_member.role) if key.team_member else TeamRole.MEMBER teams.append(TeamIdentity(id=key.team_id, name=key.team_alias, role=membership)) @@ -181,7 +181,7 @@ class DbResolver(Resolver, ProvisioningStore): ) async def _principal_from_user(self, credential: Credential, user: "LiteLLM_UserTable") -> Principal: - teams: List[TeamIdentity] = [] + teams: list[TeamIdentity] = [] for team_id in user.teams or []: try: team = await get_team_object(team_id, self._prisma, self._cache) @@ -216,7 +216,7 @@ class DbResolver(Resolver, ProvisioningStore): credential_ref=credential.credential_ref, ) - async def _organization(self, user: "LiteLLM_UserTable") -> Optional[OrganizationIdentity]: + async def _organization(self, user: "LiteLLM_UserTable") -> OrganizationIdentity | None: if user.organization_id is None: return None try: @@ -237,7 +237,7 @@ class DbResolver(Resolver, ProvisioningStore): stored = await repo.table.update(where={"user_id": user.id}, data=data) return db_user_to_scim(stored) - async def get_user(self, resource_id: str) -> Optional[ScimUser]: + async def get_user(self, resource_id: str) -> ScimUser | None: stored = await UserRepository(self._prisma).table.find_unique(where={"user_id": resource_id}) return db_user_to_scim(stored) if stored is not None else None @@ -250,7 +250,7 @@ class DbResolver(Resolver, ProvisioningStore): metadata["scim_active"] = False await repo.table.update(where={"user_id": resource_id}, data={"metadata": metadata}) - async def list_users(self, filter_expr: Optional[str]) -> List[ScimUser]: + async def list_users(self, filter_expr: str | None) -> list[ScimUser]: rows = await UserRepository(self._prisma).table.find_many() return [db_user_to_scim(row) for row in rows] @@ -264,19 +264,19 @@ class DbResolver(Resolver, ProvisioningStore): existing = await repo.find_by_id(group.id, "team_id") if group.id else None if existing is None: data["team_id"] = group.id or str(uuid.uuid4()) - stored: Optional[LiteLLM_TeamTable] = await repo.create(data) + stored: LiteLLM_TeamTable | None = await repo.create(data) else: stored = await repo.update(group.id, data, id_field="team_id") assert stored is not None return db_team_to_scim(stored) - async def get_group(self, resource_id: str) -> Optional[ScimGroup]: + async def get_group(self, resource_id: str) -> ScimGroup | None: stored = await TeamRepository(self._prisma).find_by_id(resource_id, "team_id") return db_team_to_scim(stored) if stored is not None else None async def delete_group(self, resource_id: str) -> None: await TeamRepository(self._prisma).table.delete(where={"team_id": resource_id}) - async def list_groups(self, filter_expr: Optional[str]) -> List[ScimGroup]: + async def list_groups(self, filter_expr: str | None) -> list[ScimGroup]: rows = await TeamRepository(self._prisma).find_many() return [db_team_to_scim(row) for row in rows]