(feat) /user/get_requests

litellm/proxy/proxy_server.py

@@ -4175,6 +4175,48 @@ async def user_request_model(request: Request):
         )
 
 
+@router.get(
+    "/user/get_requests",
+    tags=["user management"],
+    dependencies=[Depends(user_api_key_auth)],
+)
+async def user_get_requests():
+    """
+    Get all "Access" requests made by proxy users, access requests are requests for accessing models
+    """
+    global prisma_client
+    try:
+
+        # get the row from db
+        if prisma_client is None:
+            raise Exception("Not connected to DB!")
+
+        # TODO: Optimize this so we don't read all the data here, eventually move to pagination
+        response = await prisma_client.get_data(
+            query_type="find_all",
+            table_name="user_notification",
+        )
+        return {"requests": response}
+        # update based on remaining passed in values
+    except Exception as e:
+        traceback.print_exc()
+        if isinstance(e, HTTPException):
+            raise ProxyException(
+                message=getattr(e, "detail", f"Authentication Error({str(e)})"),
+                type="auth_error",
+                param=getattr(e, "param", "None"),
+                code=getattr(e, "status_code", status.HTTP_400_BAD_REQUEST),
+            )
+        elif isinstance(e, ProxyException):
+            raise e
+        raise ProxyException(
+            message="Authentication Error, " + str(e),
+            type="auth_error",
+            param=getattr(e, "param", "None"),
+            code=status.HTTP_400_BAD_REQUEST,
+        )
+
+
 #### TEAM MANAGEMENT ####
 
 

litellm/proxy/utils.py

@@ -512,7 +512,9 @@ class PrismaClient:
         user_id_list: Optional[list] = None,
         team_id: Optional[str] = None,
         key_val: Optional[dict] = None,
-        table_name: Optional[Literal["user", "key", "config", "spend", "team"]] = None,
+        table_name: Optional[
+            Literal["user", "key", "config", "spend", "team", "user_notification"]
+        ] = None,
         query_type: Literal["find_unique", "find_all"] = "find_unique",
         expires: Optional[datetime] = None,
         reset_at: Optional[datetime] = None,
@@ -677,6 +679,14 @@ class PrismaClient:
                         where={"members": {"has": user_id}}
                     )
                 return response
+            elif table_name == "user_notification":
+                if query_type == "find_unique":
+                    response = await self.db.litellm_usernotifications.find_unique(
+                        where={"user_id": user_id}  # type: ignore
+                    )
+                elif query_type == "find_all":
+                    response = await self.db.litellm_usernotifications.find_many()
+                return response
         except Exception as e:
             print_verbose(f"LiteLLM Prisma Client Exception: {e}")
             import traceback