Update scim.py

This commit is contained in:
Dieu 2025-07-13 23:54:33 +02:00
parent c9fda793e2
commit 1d9c1f741d

View file

@ -10,6 +10,7 @@ from typing import Optional, List, Dict, Any
from datetime import datetime, timezone from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException, Request, Query, Header, status from fastapi import APIRouter, Depends, HTTPException, Request, Query, Header, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, ConfigDict from pydantic import BaseModel, Field, ConfigDict
from open_webui.models.users import Users, UserModel from open_webui.models.users import Users, UserModel
@ -34,6 +35,29 @@ SCIM_RESOURCE_TYPE_USER = "User"
SCIM_RESOURCE_TYPE_GROUP = "Group" SCIM_RESOURCE_TYPE_GROUP = "Group"
def scim_error(status_code: int, detail: str, scim_type: Optional[str] = None):
"""Create a SCIM-compliant error response"""
error_body = {
"schemas": [SCIM_ERROR_SCHEMA],
"status": str(status_code),
"detail": detail
}
if scim_type:
error_body["scimType"] = scim_type
elif status_code == 404:
error_body["scimType"] = "invalidValue"
elif status_code == 409:
error_body["scimType"] = "uniqueness"
elif status_code == 400:
error_body["scimType"] = "invalidSyntax"
return JSONResponse(
status_code=status_code,
content=error_body
)
class SCIMError(BaseModel): class SCIMError(BaseModel):
"""SCIM Error Response""" """SCIM Error Response"""
schemas: List[str] = [SCIM_ERROR_SCHEMA] schemas: List[str] = [SCIM_ERROR_SCHEMA]
@ -249,6 +273,7 @@ def get_scim_auth(request: Request, authorization: Optional[str] = Header(None))
) )
def user_to_scim(user: UserModel, request: Request) -> SCIMUser: def user_to_scim(user: UserModel, request: Request) -> SCIMUser:
"""Convert internal User model to SCIM User""" """Convert internal User model to SCIM User"""
# Parse display name into name components # Parse display name into name components
@ -493,9 +518,9 @@ async def get_user(
"""Get SCIM User by ID""" """Get SCIM User by ID"""
user = Users.get_user_by_id(user_id) user = Users.get_user_by_id(user_id)
if not user: if not user:
raise HTTPException( return scim_error(
status_code=status.HTTP_404_NOT_FOUND, status_code=status.HTTP_404_NOT_FOUND,
detail=f"User {user_id} not found", detail=f"User {user_id} not found"
) )
return user_to_scim(user, request) return user_to_scim(user, request)