refactor scim transform

litellm/proxy/management_endpoints/scim/README_SCIM.md

@@ -68,13 +68,6 @@ Most identity providers will require authentication. You should use a valid Lite
 - Automatic synchronization of user-team relationships
 - Proper status codes and error handling per SCIM specification
 
-## Limitations
-
-This is a basic implementation intended for testing and integration purposes. Some limitations include:
-
-- Limited filtering capabilities (only supports basic equality filters)
-- No support for complex SCIM operations like PATCH
-- Limited schema customization
 
 ## Example Usage
 

litellm/proxy/management_endpoints/scim/scim_transformations.py

@@ -0,0 +1,154 @@
+from typing import List, Union
+
+from litellm.proxy._types import (
+    LiteLLM_TeamTable,
+    LiteLLM_UserTable,
+    Member,
+    NewUserResponse,
+)
+from litellm.types.proxy.management_endpoints.scim_v2 import *
+
+
+class ScimTransformations:
+    DEFAULT_SCIM_NAME = "Unknown User"
+    DEFAULT_SCIM_FAMILY_NAME = "Unknown Family Name"
+    DEFAULT_SCIM_DISPLAY_NAME = "Unknown Display Name"
+    DEFAULT_SCIM_MEMBER_VALUE = "Unknown Member Value"
+
+    @staticmethod
+    async def transform_litellm_user_to_scim_user(
+        user: Union[LiteLLM_UserTable, NewUserResponse],
+    ) -> SCIMUser:
+        from litellm.proxy.proxy_server import prisma_client
+
+        if prisma_client is None:
+            raise HTTPException(
+                status_code=500, detail={"error": "No database connected"}
+            )
+
+        # Get user's teams/groups
+        groups = []
+        for team_id in user.teams or []:
+            team = await prisma_client.db.litellm_teamtable.find_unique(
+                where={"team_id": team_id}
+            )
+            if team:
+                team_alias = getattr(team, "team_alias", team.team_id)
+                groups.append(SCIMUserGroup(value=team.team_id, display=team_alias))
+
+        user_created_at = user.created_at.isoformat() if user.created_at else None
+        user_updated_at = user.updated_at.isoformat() if user.updated_at else None
+
+        emails = []
+        if user.user_email:
+            emails.append(SCIMUserEmail(value=user.user_email, primary=True))
+
+        return SCIMUser(
+            schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
+            id=user.user_id,
+            userName=ScimTransformations._get_scim_user_name(user),
+            displayName=ScimTransformations._get_scim_user_name(user),
+            name=SCIMUserName(
+                familyName=ScimTransformations._get_scim_family_name(user),
+                givenName=ScimTransformations._get_scim_given_name(user),
+            ),
+            emails=emails,
+            groups=groups,
+            active=True,
+            meta={
+                "resourceType": "User",
+                "created": user_created_at,
+                "lastModified": user_updated_at,
+            },
+        )
+
+    @staticmethod
+    def _get_scim_user_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str:
+        """
+        SCIM requires a display name with length > 0
+
+        We use the same userName and displayName for SCIM users
+        """
+        if user.user_email and len(user.user_email) > 0:
+            return user.user_email
+        return ScimTransformations.DEFAULT_SCIM_DISPLAY_NAME
+
+    @staticmethod
+    def _get_scim_family_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str:
+        """
+        SCIM requires a family name with length > 0
+        """
+        metadata = user.metadata or {}
+        if "scim_metadata" in metadata:
+            scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata(
+                **metadata["scim_metadata"]
+            )
+            if scim_metadata.familyName and len(scim_metadata.familyName) > 0:
+                return scim_metadata.familyName
+
+        if user.user_alias and len(user.user_alias) > 0:
+            return user.user_alias
+        return ScimTransformations.DEFAULT_SCIM_FAMILY_NAME
+
+    @staticmethod
+    def _get_scim_given_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str:
+        """
+        SCIM requires a given name with length > 0
+        """
+        metadata = user.metadata or {}
+        if "scim_metadata" in metadata:
+            scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata(
+                **metadata["scim_metadata"]
+            )
+            if scim_metadata.givenName and len(scim_metadata.givenName) > 0:
+                return scim_metadata.givenName
+
+        if user.user_alias and len(user.user_alias) > 0:
+            return user.user_alias or ScimTransformations.DEFAULT_SCIM_NAME
+        return ScimTransformations.DEFAULT_SCIM_NAME
+
+    @staticmethod
+    async def transform_litellm_team_to_scim_group(
+        team: Union[LiteLLM_TeamTable, dict],
+    ) -> SCIMGroup:
+        from litellm.proxy.proxy_server import prisma_client
+
+        if prisma_client is None:
+            raise HTTPException(
+                status_code=500, detail={"error": "No database connected"}
+            )
+
+        if isinstance(team, dict):
+            team = LiteLLM_TeamTable(**team)
+
+        # Get team members
+        scim_members: List[SCIMMember] = []
+        for member in team.members_with_roles or []:
+            scim_members.append(
+                SCIMMember(
+                    value=ScimTransformations._get_scim_member_value(member),
+                    display=member.user_email,
+                )
+            )
+
+        team_alias = getattr(team, "team_alias", team.team_id)
+        team_created_at = team.created_at.isoformat() if team.created_at else None
+        team_updated_at = team.updated_at.isoformat() if team.updated_at else None
+
+        return SCIMGroup(
+            schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
+            id=team.team_id,
+            displayName=team_alias,
+            members=scim_members,
+            meta={
+                "resourceType": "Group",
+                "created": team_created_at,
+                "lastModified": team_updated_at,
+            },
+        )
+
+    @staticmethod
+    def _get_scim_member_value(member: Member) -> str:
+        if member.user_email:
+            return member.user_email
+        return ScimTransformations.DEFAULT_SCIM_MEMBER_VALUE

litellm/proxy/management_endpoints/scim/scim_v2.py

@@ -1,12 +1,10 @@
 """
 SCIM v2 Endpoints for LiteLLM Proxy using Internal User/Team Management
 
-Provides basic implementations of SCIM v2 endpoints for testing
-and integration purposes.
 """
 
 import uuid
-from typing import List, Optional, Union
+from typing import List, Optional
 
 from fastapi import (
     APIRouter,
@@ -21,17 +19,18 @@ from fastapi import (
 
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy._types import (
-    LiteLLM_TeamTable,
     LiteLLM_UserTable,
     LitellmUserRoles,
     Member,
     NewTeamRequest,
     NewUserRequest,
-    NewUserResponse,
     UserAPIKeyAuth,
 )
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.management_endpoints.internal_user_endpoints import new_user
+from litellm.proxy.management_endpoints.scim.scim_transformations import (
+    ScimTransformations,
+)
 from litellm.proxy.management_endpoints.team_endpoints import new_team
 from litellm.types.proxy.management_endpoints.scim_v2 import *
 
@@ -49,151 +48,6 @@ async def set_scim_content_type(response: Response):
     response.headers["Content-Type"] = "application/scim+json"
 
 
-class ScimTransformations:
-    DEFAULT_SCIM_NAME = "Unknown User"
-    DEFAULT_SCIM_FAMILY_NAME = "Unknown Family Name"
-    DEFAULT_SCIM_DISPLAY_NAME = "Unknown Display Name"
-    DEFAULT_SCIM_MEMBER_VALUE = "Unknown Member Value"
-
-    @staticmethod
-    async def transform_litellm_user_to_scim_user(
-        user: Union[LiteLLM_UserTable, NewUserResponse],
-    ) -> SCIMUser:
-        from litellm.proxy.proxy_server import prisma_client
-
-        if prisma_client is None:
-            raise HTTPException(
-                status_code=500, detail={"error": "No database connected"}
-            )
-
-        # Get user's teams/groups
-        groups = []
-        for team_id in user.teams or []:
-            team = await prisma_client.db.litellm_teamtable.find_unique(
-                where={"team_id": team_id}
-            )
-            if team:
-                team_alias = getattr(team, "team_alias", team.team_id)
-                groups.append(SCIMUserGroup(value=team.team_id, display=team_alias))
-
-        user_created_at = user.created_at.isoformat() if user.created_at else None
-        user_updated_at = user.updated_at.isoformat() if user.updated_at else None
-
-        emails = []
-        if user.user_email:
-            emails.append(SCIMUserEmail(value=user.user_email, primary=True))
-
-        return SCIMUser(
-            schemas=["urn:ietf:params:scim:schemas:core:2.0:User"],
-            id=user.user_id,
-            userName=ScimTransformations._get_scim_user_name(user),
-            displayName=ScimTransformations._get_scim_user_name(user),
-            name=SCIMUserName(
-                familyName=ScimTransformations._get_scim_family_name(user),
-                givenName=ScimTransformations._get_scim_given_name(user),
-            ),
-            emails=emails,
-            groups=groups,
-            active=True,
-            meta={
-                "resourceType": "User",
-                "created": user_created_at,
-                "lastModified": user_updated_at,
-            },
-        )
-
-    @staticmethod
-    def _get_scim_user_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str:
-        """
-        SCIM requires a display name with length > 0
-
-        We use the same userName and displayName for SCIM users
-        """
-        if user.user_email and len(user.user_email) > 0:
-            return user.user_email
-        return ScimTransformations.DEFAULT_SCIM_DISPLAY_NAME
-
-    @staticmethod
-    def _get_scim_family_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str:
-        """
-        SCIM requires a family name with length > 0
-        """
-        metadata = user.metadata or {}
-        if "scim_metadata" in metadata:
-            scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata(
-                **metadata["scim_metadata"]
-            )
-            if scim_metadata.familyName and len(scim_metadata.familyName) > 0:
-                return scim_metadata.familyName
-
-        if user.user_alias and len(user.user_alias) > 0:
-            return user.user_alias
-        return ScimTransformations.DEFAULT_SCIM_FAMILY_NAME
-
-    @staticmethod
-    def _get_scim_given_name(user: Union[LiteLLM_UserTable, NewUserResponse]) -> str:
-        """
-        SCIM requires a given name with length > 0
-        """
-        metadata = user.metadata or {}
-        if "scim_metadata" in metadata:
-            scim_metadata: LiteLLM_UserScimMetadata = LiteLLM_UserScimMetadata(
-                **metadata["scim_metadata"]
-            )
-            if scim_metadata.givenName and len(scim_metadata.givenName) > 0:
-                return scim_metadata.givenName
-
-        if user.user_alias and len(user.user_alias) > 0:
-            return user.user_alias or ScimTransformations.DEFAULT_SCIM_NAME
-        return ScimTransformations.DEFAULT_SCIM_NAME
-
-    @staticmethod
-    async def transform_litellm_team_to_scim_group(
-        team: Union[LiteLLM_TeamTable, dict],
-    ) -> SCIMGroup:
-        from litellm.proxy.proxy_server import prisma_client
-
-        if prisma_client is None:
-            raise HTTPException(
-                status_code=500, detail={"error": "No database connected"}
-            )
-
-        if isinstance(team, dict):
-            team = LiteLLM_TeamTable(**team)
-
-        # Get team members
-        scim_members: List[SCIMMember] = []
-        for member in team.members_with_roles or []:
-            scim_members.append(
-                SCIMMember(
-                    value=ScimTransformations._get_scim_member_value(member),
-                    display=member.user_email,
-                )
-            )
-
-        team_alias = getattr(team, "team_alias", team.team_id)
-        team_created_at = team.created_at.isoformat() if team.created_at else None
-        team_updated_at = team.updated_at.isoformat() if team.updated_at else None
-
-        return SCIMGroup(
-            schemas=["urn:ietf:params:scim:schemas:core:2.0:Group"],
-            id=team.team_id,
-            displayName=team_alias,
-            members=scim_members,
-            meta={
-                "resourceType": "Group",
-                "created": team_created_at,
-                "lastModified": team_updated_at,
-            },
-        )
-
-    @staticmethod
-    def _get_scim_member_value(member: Member) -> str:
-        if member.user_email:
-            return member.user_email
-        return ScimTransformations.DEFAULT_SCIM_MEMBER_VALUE
-
-
 # User Endpoints
 @scim_router.get(
     "/Users",