fix(managed_files): skip unparseable rows when listing managed files

enterprise/litellm_enterprise/proxy/hooks/managed_files.py

@@ -73,6 +73,20 @@
     PrismaClient = Any
 
 
+def _parse_managed_file_object(
+    raw_file_object: object, unified_file_id: str
+) -> Optional[OpenAIFileObject]:
+    if raw_file_object is None:
+        return None
+    try:
+        return OpenAIFileObject.model_validate(raw_file_object)
+    except Exception as e:
+        verbose_logger.warning(
+            f"Failed to parse managed file object {unified_file_id}: {e}"
+        )
+        return None
+
+
 class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
     # Class variables or attributes
     def __init__(
@@ -383,9 +397,14 @@ async def get_user_created_file_ids(
             }
         )
         return [
-            OpenAIFileObject.model_validate(file_object.file_object)
+            parsed_file_object
             for file_object in file_ids
-            if file_object.file_object is not None
+            if (
+                parsed_file_object := _parse_managed_file_object(
+                    file_object.file_object, file_object.unified_file_id
+                )
+            )
+            is not None
         ]
 
     async def check_managed_file_id_access(

tests/test_litellm/enterprise/proxy/test_managed_files_hook.py

@@ -154,6 +154,29 @@ async def test_get_user_created_file_ids_skips_rows_without_file_object():
     assert [file.id for file in files] == ["file-output-abc"]
 
 
+@pytest.mark.asyncio
+async def test_get_user_created_file_ids_skips_unparseable_rows():
+    managed_files = _make_managed_files_instance()
+    managed_files.prisma_client.db.litellm_managedfiletable.find_many = AsyncMock(
+        return_value=[
+            MagicMock(
+                file_object={"id": "file-corrupt", "object": "file"},
+                unified_file_id="unified-corrupt",
+            ),
+            MagicMock(
+                file_object=_make_file_object().model_dump(),
+                unified_file_id="unified-valid",
+            ),
+        ]
+    )
+
+    files = await managed_files.get_user_created_file_ids(
+        _make_user_api_key_dict(), ["file-output-abc"]
+    )
+
+    assert [file.id for file in files] == ["file-output-abc"]
+
+
 @pytest.mark.asyncio
 async def test_should_fallback_when_no_router():
     """