refactor(vector-stores): extract _fetch_and_authorize_vector_store helper

/simplify pass:
- ``update_vector_store`` (newly added) and ``get_vector_store_info``'s
  DB-fallback path duplicated the same shape: ``find_unique`` →
  ``model_dump`` → ``LiteLLM_ManagedVectorStore(**)`` →
  ``_check_vector_store_access`` → raise 404/403. Extract into
  ``_fetch_and_authorize_vector_store`` so the pattern lives in one
  place; future endpoints that need the same gate get it via one call.
- The ``except HTTPException: raise`` guard added in the prior commit is
  retained — the helper raises HTTPException(403/404) and the catch-all
  ``except Exception`` would otherwise rewrite them as 500.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
user 2026-04-25 05:14:34 +00:00
parent 51d560ba2e
commit 78d12ee888
No known key found for this signature in database

View File

@ -63,6 +63,33 @@ def _redact_sensitive_litellm_params(
}
async def _fetch_and_authorize_vector_store(
vector_store_id: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: Any,
) -> "LiteLLM_ManagedVectorStore":
"""
Look up a vector store by id and confirm the caller can access it.
Raises HTTPException(404) on miss and HTTPException(403) on access
denial.
"""
row = await prisma_client.db.litellm_managedvectorstorestable.find_unique(
where={"vector_store_id": vector_store_id}
)
if row is None:
raise HTTPException(
status_code=404,
detail=f"Vector store with ID {vector_store_id} not found",
)
typed = LiteLLM_ManagedVectorStore(**row.model_dump())
if not await _check_vector_store_access(typed, user_api_key_dict):
raise HTTPException(
status_code=403,
detail="Access denied: You do not have permission to access this vector store",
)
return typed
def _resolve_embedding_config_from_router(
embedding_model: str, llm_router
) -> Optional[Dict[str, Any]]:
@ -752,26 +779,12 @@ async def get_vector_store_info(
)
return {"vector_store": vector_store_pydantic_obj}
vector_store = (
await prisma_client.db.litellm_managedvectorstorestable.find_unique(
where={"vector_store_id": data.vector_store_id}
)
vector_store_typed = await _fetch_and_authorize_vector_store(
vector_store_id=data.vector_store_id,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
if vector_store is None:
raise HTTPException(
status_code=404,
detail=f"Vector store with ID {data.vector_store_id} not found",
)
# Check access control for DB vector store
vector_store_dict = vector_store.model_dump() # type: ignore[attr-defined]
vector_store_typed = LiteLLM_ManagedVectorStore(**vector_store_dict)
if not await _check_vector_store_access(vector_store_typed, user_api_key_dict):
raise HTTPException(
status_code=403,
detail="Access denied: You do not have permission to access this vector store",
)
vector_store_dict = dict(vector_store_typed)
if "litellm_params" in vector_store_dict:
vector_store_dict["litellm_params"] = _redact_sensitive_litellm_params(
vector_store_dict["litellm_params"]
@ -809,22 +822,12 @@ async def update_vector_store(
# Per-store access control: anyone authenticated who passes the
# premium-feature gate could otherwise update *any* vector store —
# including stores belonging to other teams. Mirror the check
# ``/vector_store/info`` already performs.
existing = await prisma_client.db.litellm_managedvectorstorestable.find_unique(
where={"vector_store_id": vector_store_id}
# including stores belonging to other teams.
await _fetch_and_authorize_vector_store(
vector_store_id=vector_store_id,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
if existing is None:
raise HTTPException(
status_code=404,
detail=f"Vector store with ID {vector_store_id} not found",
)
existing_typed = LiteLLM_ManagedVectorStore(**existing.model_dump())
if not await _check_vector_store_access(existing_typed, user_api_key_dict):
raise HTTPException(
status_code=403,
detail="Access denied: You do not have permission to update this vector store",
)
# Handle metadata serialization
if update_data.get("vector_store_metadata") is not None: