"""Shared FastAPI dependencies for bulk-session routes.""" from typing import Any from uuid import UUID from fastapi import Depends, HTTPException, Path from fastapi.requests import Request from pydantic import UUID4 from product_staging.api.auth import identity_uuid_from_scope, profiles_from_scope from product_staging.api.schemas.bulk_session import BulkSessionId from product_staging.constants.error import ERROR_MESSAGE_NO_BULK_SESSION from product_staging.logic import bulk_session as bulk_session_logic SLUG_REDIRECT_RESPONSES: dict[int | str, dict[str, Any]] = { 307: {"description": "Redirect from a slug to the canonical bulk-session UUID URL."} } class BulkSessionSlugRedirect(Exception): """Raised to 307-redirect a slug request to its canonical UUID URL. Carries the relative ``Location`` the global handler should redirect to. """ def __init__(self, location: str) -> None: self.location = location async def resolve_bulk_session_slug( request: Request, bulk_session_id: BulkSessionId = Path(description="Bulk session UUID or slug"), identity_uuid: UUID4 = Depends(identity_uuid_from_scope), profiles: list[tuple[str, int]] = Depends(profiles_from_scope), ) -> UUID: """Resolve a ``bulk_session_id`` path param that may be a UUID or a slug. A UUID is returned unchanged. A slug ('{vendor_id}-YYYY-MM-DD-HH-MM-SS') is resolved to its canonical UUID and, after authorization, a :class:`BulkSessionSlugRedirect` is raised so the request is answered with a 307 to the canonical UUID URL (relative Location, query string preserved). """ if isinstance(bulk_session_id, UUID): return bulk_session_id resolved_id = await bulk_session_logic.get_bulk_session_id_by_slug(bulk_session_id) if not resolved_id: raise HTTPException(status_code=404, detail=ERROR_MESSAGE_NO_BULK_SESSION) resolved_uuid = UUID(resolved_id) await bulk_session_logic.assert_authorization( bulk_session_id=resolved_uuid, identity_uuid=identity_uuid, profiles=profiles, ) new_path = request.url.path.replace(bulk_session_id, str(resolved_uuid)) location = f"{new_path}?{request.url.query}" if request.url.query else new_path raise BulkSessionSlugRedirect(location)