"""Bulk Session CRUD operation.""" import json import logging import re import uuid from collections import defaultdict from datetime import datetime, timedelta, timezone from enum import Enum from typing import Any, Dict, cast from uuid import UUID from ddtrace.trace import tracer from fastapi import HTTPException from pydantic import UUID1, UUID4 from product_staging import config from product_staging.api import datasources from product_staging.api.auth import ( TenantLookup, check_vendor_access_for_profiles, get_tenant, get_tenants_many, is_authorized_for_tenant, is_authorized_for_tenants, ) from product_staging.api.schemas.bulk_session import ( AssetStatus, CreateBulkSessionIngestionRequest, MetadataStatus, UpdateBulkSessionRequest, ) from product_staging.api.schemas.bulk_session_asset_file import FileStatus from product_staging.api.schemas.metadata_error_report import ( MetadataErrorReport, MetadataErrorReportResponse, ) from product_staging.api.schemas.tenant import Tenant from product_staging.connectors import db, ows_account from product_staging.connectors.features.wrappers import BooleanFeature from product_staging.constants import header from product_staging.constants.bulk_session_ingestion import IngestionStatus from product_staging.constants.error import ( ERROR_MESSAGE_NO_ACCOUNT_FOUND, ERROR_MESSAGE_NO_BULK_SESSION, ERROR_MESSAGE_NO_ERROR_REPORT, ) from product_staging.constants.features import FEATURE_FLAG_CCM_SESSION_INGEST_2026 from product_staging.logic import ( bulk_session_ingestion_execution as bulk_session_ingestion_execution_logic, ) from product_staging.logic import bulk_session_metadata_file from product_staging.logic import metadata_upload as metadata_upload_logic from product_staging.logic.utils import aws_lambda, s3, utils from product_staging.models import ( bulk_session as bulk_session_model, ) from product_staging.models import ( bulk_session_asset_cloud_transfer_job as bulk_session_asset_cloud_transfer_job_model, ) from product_staging.models import ( bulk_session_asset_file, bulk_session_ingestion, bulk_session_ingestion_product, bulk_session_ingestion_project, bulk_session_ingestion_track, ) from product_staging.schema.metadata_json import ProductObject logger = logging.getLogger(__name__) async def _require_vendor_id(vendor_uuid: str | UUID1 | UUID4) -> int: """Resolve and require numeric vendor_id for API response payloads.""" resolved_vendor_id = await ows_account.get_vendor_id(UUID(str(vendor_uuid))) if resolved_vendor_id is None: raise HTTPException( status_code=500, detail="Failed to resolve vendor_id for bulk session vendor_uuid", ) return resolved_vendor_id class SortField(str, Enum): """Available fields for sorting both products and projects.""" NAME = "name" ERRORS = "errors" class SortOrder(str, Enum): """Sort order options.""" ASC = "asc" DESC = "desc" _num = re.compile(r"(\d+)") def natural_key(s: str): s = s or "" parts = _num.split(s) return tuple((0, int(p)) if p.isdigit() else (1, p.casefold()) for p in parts if p) @db.db_session_wrap async def create_bulk_session( vendor_uuid: UUID1 | UUID4 | None, identity_uuid: UUID4, subaccount_id: int | None, session=None, ): """Create a bulk_session object.""" if subaccount_id and not vendor_uuid: vendor_result = await ows_account.get_vendor_uuid_from_subaccount_id( subaccount_id ) if vendor_result: vendor_uuid = UUID(vendor_result) if not vendor_uuid: raise HTTPException(status_code=404, detail=ERROR_MESSAGE_NO_ACCOUNT_FOUND) created_bulk_session = await bulk_session_model.create_bulk_session( vendor_uuid=vendor_uuid, identity_uuid=identity_uuid, subaccount_id=subaccount_id, session=session, ) created_bulk_session.vendor_id = await _require_vendor_id( created_bulk_session.vendor_uuid ) return created_bulk_session @db.db_session_wrap async def get_bulk_session( bulk_session_id: UUID4, identity_uuid: UUID4, get_download_link=True, is_cancelled: bool | None = False, enrich=True, resolve_slug: bool = False, session=None, ): """Retrieve bulk session. is_cancelled parameter is used to filter out "cancelled" sessions by default: to reset the filter, pass is_cancelled=None; to get only cancelled sessions pass is_cancelled=True. resolve_slug resolves the numeric vendor id (a network hop to ows-account) so the slug can be built; off by default to avoid that hop for callers that don't need it. """ bulk_session = await bulk_session_model.get_bulk_session( bulk_session_id, is_cancelled=is_cancelled, session=session ) if not bulk_session: raise HTTPException(status_code=404, detail=ERROR_MESSAGE_NO_BULK_SESSION) if enrich: bulk_session = await _enrich_bulk_session( bulk_session, identity_uuid, get_download_link=get_download_link ) if resolve_slug: bulk_session.vendor_id = await ows_account.get_vendor_id( UUID(bulk_session.vendor_uuid) ) return bulk_session async def get_bulk_session_id_by_slug(slug: str) -> str | None: """Resolve a slug to its bulk_session_id (UUID string), or None if not found.""" vendor_id_str, _, created_on_str = slug.partition("-") created_on = datetime.strptime(created_on_str, "%Y-%m-%d-%H-%M-%S") vendor_uuid = await ows_account.get_vendor_uuid(int(vendor_id_str)) if not vendor_uuid: return None bulk_session = await bulk_session_model.get_bulk_session_by_vendor_and_created_on( vendor_uuid, created_on ) return bulk_session.bulk_session_id if bulk_session else None @db.db_session_wrap async def get_bulk_sessions( bulk_session_ids: list[UUID4], is_cancelled: bool | None = False, session=None, ): """Retrieve bulk sessions by bulk_session_ids. is_cancelled parameter is forwarded to the model layer: pass is_cancelled=None to return all sessions regardless of cancellation state, or is_cancelled=True to return only cancelled sessions. """ return await bulk_session_model.get_bulk_sessions( bulk_session_ids, is_cancelled=is_cancelled, session=session ) async def get_session_metadata( bulk_session_data: bulk_session_model.BulkSession, ) -> list[ProductObject] | None: """ Given a bulk session, read its metadata file from S3 and return as a list of ProductObjects. If the bulk session does not have a validated metadata file, return None. Returns: List of ProductObject instances parsed from the metadata JSON file, or None if no validated metadata file is present. """ if not await bulk_session_data.validated_metadata_json_file(): return None metadata_json_file = await s3.get_file( await bulk_session_data.validated_metadata_json_file() ) return json.loads((await metadata_json_file["Body"].read())) @db.db_session_wrap async def update_bulk_session( bulk_session_id: UUID4, identity_uuid: UUID4, update: UpdateBulkSessionRequest, session=None, ): """Update bulk session.""" updated_bulk_session = await bulk_session_model.update_bulk_session( bulk_session_id, identity_uuid, update, session=session ) updated_bulk_session.vendor_id = await _require_vendor_id( updated_bulk_session.vendor_uuid ) return updated_bulk_session @db.db_session_wrap async def create_bulk_session_ingestion( bulk_session_id: UUID4, identity_uuid: UUID4, ingest: CreateBulkSessionIngestionRequest, session=None, ): """Create bulk session ingestion and add it to the queue.""" bulk_session = await get_bulk_session( bulk_session_id, identity_uuid, session=session ) json_file_name = f"metadata_ingestion/{uuid.uuid4()}" metadata = await get_session_metadata(bulk_session) if not metadata: raise HTTPException( status_code=400, detail="Failed to read metadata from bulk session's validated metadata file", ) filtered_metadata = bulk_session_metadata_file.filter_metadata( metadata, assets_required=ingest.assets_required ) await s3.write_file_stream( json_file_name, utils.list_to_stream(filtered_metadata), metadata={"bulk_session_id": str(bulk_session.bulk_session_id)}, ) ingestion = await bulk_session_ingestion.create_bulk_session_ingestion( bulk_session_id=bulk_session_id, assets_required=ingest.assets_required, submit_products=ingest.submit, send_notifications=ingest.send_notifications, identity_uuid=identity_uuid, json_file_name=json_file_name, total_products=len(filtered_metadata), ) if not ingestion: raise HTTPException( status_code=500, detail="Failed to create bulk session ingestion record" ) bulk_session_ingestion_id = ingestion["bulk_session_ingestion_id"] sqs_client = datasources.get_sqs_client() session_ids = f"{bulk_session_id}___{bulk_session_ingestion_id}" message_id = session_ids + "___" + str(datetime.now().strftime("%Y%m%d%H%M%S")) sqs_client.send_message( QueueUrl=config.INGESTION_SQS_QUEUE_URL, MessageGroupId=config.INGESTION_SQS_QUEUE_MESSAGE_GROUP_ID, MessageDeduplicationId=message_id, MessageBody=message_id, MessageAttributes={ "BulkSessionId": { "DataType": "String", "StringValue": str(bulk_session_id), }, "BulkSessionIngestionId": { "DataType": "String", "StringValue": str(bulk_session_ingestion_id), }, }, ) aws_lambda.invoke( function_name=config.TRIGGER_INGEST_LAMBDA_ARN, payload={}, ) return ingestion async def read_ingestion_message_from_sqs() -> Dict[str, Any] | None: sqs_client = datasources.get_sqs_client() response = sqs_client.receive_message( QueueUrl=config.INGESTION_SQS_QUEUE_URL, MaxNumberOfMessages=1, VisibilityTimeout=config.INGESTION_SQS_QUEUE_MESSAGE_PROCESSING_TIME, WaitTimeSeconds=config.INGESTION_SQS_QUEUE_MESSAGE_PROCESSING_TIME + 1, MessageAttributeNames=["All"], ) if not response or not response.get("Messages"): return None return response["Messages"][0] async def delete_ingestion_message_from_sqs(sqs_message: Dict[str, Any]) -> None: sqs_client = datasources.get_sqs_client() sqs_client.delete_message( QueueUrl=config.INGESTION_SQS_QUEUE_URL, ReceiptHandle=sqs_message["ReceiptHandle"], ) async def ingest_bulk_session(*, sqs_message: Dict[str, Any], identity_uuid: UUID4): message_attributes = sqs_message["MessageAttributes"] bulk_session_ingestion_id = message_attributes["BulkSessionIngestionId"][ "StringValue" ] ingestion = await bulk_session_ingestion.get_bulk_session_ingestion( bulk_session_ingestion_id ) asset_queue_feature = BooleanFeature( client=datasources.get_splitio_client(), feature_name=FEATURE_FLAG_CCM_SESSION_INGEST_2026, ) use_asset_queue = asset_queue_feature.is_on_for_identity( ingestion.get("identity_uuid") ) state_machine_arn = ( config.BULK_SESSION_INGEST_QUEUE_SFN_ARN if use_asset_queue else config.BULK_SESSION_INGEST_SFN_ARN ) max_concurrency = config.BULK_SESSION_INGEST_QUEUE_MAX_CONCURRENCY sfn_client = datasources.get_sfn_client() response = sfn_client.list_executions( stateMachineArn=state_machine_arn, statusFilter="RUNNING", maxResults=max_concurrency, ) execution_list = response and response.get("executions") if use_asset_queue and execution_list and len(execution_list) >= max_concurrency: return None if not use_asset_queue and execution_list: return None bulk_session_id = ingestion["bulk_session_id"] response = sfn_client.start_execution( stateMachineArn=state_machine_arn, name=str(bulk_session_id) + "_" + str(datetime.now().strftime("%Y%m%d%H%M%S")), input=json.dumps( { "bulk_session_id": bulk_session_id, "bulk_session_ingestion_id": bulk_session_ingestion_id, "identity_uuid": str(ingestion["created_by"]), "assets_required": ingestion["assets_required"], "submit_products": ingestion["submit_products"], "send_notifications": ingestion["send_notifications"], "ingestion_status": "in_progress", "correlation_id": str(uuid.uuid4()), "bucket": config.OWS_PRODUCT_STAGING_S3_BUCKET, "key": ingestion["json_file_name"], } ), ) response_status = response["ResponseMetadata"]["HTTPStatusCode"] if response_status != 200: return response_status await delete_ingestion_message_from_sqs(sqs_message) execution_arn = response["executionArn"] await ( bulk_session_ingestion_execution_logic.upsert_bulk_session_ingestion_execution( bulk_session_ingestion_id=bulk_session_ingestion_id, execution_arn=execution_arn, identity_uuid=identity_uuid, ) ) return { "execution_arn": execution_arn, "start_date": response["startDate"], } @db.db_session_wrap async def get_metadata_error_report( bulk_session_id: UUID4, limit: int, project_offset: int, product_offset: int, product_sort_by: SortField | None = None, product_sort_order: SortOrder = SortOrder.ASC, project_sort_by: SortField | None = None, project_sort_order: SortOrder = SortOrder.ASC, session=None, ) -> MetadataErrorReportResponse: """ Returns a MetadataErrorReport for the provided bulk session id. The report is derived from the metadata error report json for the bulk session on s3. """ bulk_session = await bulk_session_model.get_bulk_session( bulk_session_id=bulk_session_id, session=session ) if not bulk_session: raise HTTPException(status_code=404, detail=ERROR_MESSAGE_NO_BULK_SESSION) if not bulk_session.metadata_error_report_json: raise HTTPException(status_code=404, detail=ERROR_MESSAGE_NO_ERROR_REPORT) error_report_file = await s3.get_file(bulk_session.metadata_error_report_json) error_report_dict = json.loads((await error_report_file["Body"].read())) error_report = MetadataErrorReport.model_validate(error_report_dict) num_products = len(error_report.products) num_projects = len(error_report.projects) if product_sort_by: error_report.products = _sort_items( error_report.products, product_sort_by, product_sort_order, _get_sort_key ) if project_sort_by: error_report.projects = _sort_items( error_report.projects, project_sort_by, project_sort_order, _get_sort_key ) error_report.products = error_report.products[ product_offset : (limit + product_offset) ] error_report.projects = error_report.projects[ project_offset : (limit + project_offset) ] return MetadataErrorReportResponse( metadata_error_report=error_report, total_projects=num_projects, total_products=num_products, ) @db.db_session_wrap async def get_created_bulk_sessions( identity_uuid: UUID4, limit: int = 10, offset: int = 0, session=None ): """Retrieve bulk sessions that were created by the given identity.""" created_sessions = await bulk_session_model.get_created_bulk_sessions( identity_uuid, limit=limit, offset=offset, session=session ) vendor_id_by_uuid: dict[str, int] = {} for created_session in created_sessions: vendor_uuid = str(created_session["vendor_uuid"]) if vendor_uuid not in vendor_id_by_uuid: vendor_id_by_uuid[vendor_uuid] = await _require_vendor_id(vendor_uuid) created_session["vendor_id"] = vendor_id_by_uuid[vendor_uuid] return created_sessions async def has_created_bulk_sessions(identity_uuid: UUID4) -> bool: """Check if the identity has created at least one bulk session.""" return await bulk_session_model.has_created_bulk_sessions(identity_uuid) async def _enrich_bulk_session( bulk_session: bulk_session_model.BulkSession, identity_uuid: UUID4, get_download_link=False, update_asset_status=True, ): """Populate additional data on a bulk session object.""" bulk_session_id = bulk_session.bulk_session_id if ( datetime.now(timezone.utc) - bulk_session.updated_on.replace(tzinfo=timezone.utc) > timedelta(hours=1) and bulk_session.metadata_status == MetadataStatus.uploading ): bulk_session.metadata_status = MetadataStatus.upload_timeout await bulk_session.latest_upload_file_id() await bulk_session.latest_upload_original_filename() download_link = None if get_download_link: if bulk_session.metadata_status == MetadataStatus.valid: download_link = await metadata_upload_logic.get_download_link( bulk_session_id, await bulk_session.success_file_id() ) elif ( bulk_session.metadata_status == MetadataStatus.invalid and await bulk_session.failure_file_id() ): download_link = await metadata_upload_logic.get_download_link( bulk_session_id, await bulk_session.failure_file_id() ) bulk_session.download_link = download_link await bulk_session.validated_metadata_json_file() if bulk_session.asset_status == AssetStatus.incomplete and update_asset_status: if await bulk_session.validated_metadata_json_file(): bulk_session.asset_status = await refresh_bulk_session_asset_status( bulk_session_id, identity_uuid, await bulk_session.validated_metadata_json_file(), ) bulk_session.latest_cloud_transfer_job_id = await bulk_session_asset_cloud_transfer_job_model.get_latest_transfer_job_id_by_bulk_session_id( bulk_session_id ) return bulk_session async def refresh_bulk_session_asset_status( bulk_session_id: UUID4, identity_uuid: UUID4, json_file_name: str, ): session_assets = await bulk_session_asset_file.get_bulk_session_assets( bulk_session_id ) valid_assets = set() for asset in session_assets: if asset.file_status == FileStatus.success.value: valid_assets.add(asset.s3_filename) if not valid_assets: updated_asset_status = AssetStatus.incomplete else: updated_asset_status = AssetStatus.complete metadata_json_file = await s3.get_file(json_file_name) metadata_json = json.loads((await metadata_json_file["Body"].read())) for product in metadata_json: artwork_key = None artwork = product["product_info"]["product"]["artwork"] if artwork: artwork_key = artwork["key"] if not artwork_key or artwork_key not in valid_assets: updated_asset_status = AssetStatus.incomplete break for track in product["product_info"]["tracks"]: audio_key = None audio = track["asset"] if audio: audio_key = audio["key"] if not audio_key or audio_key not in valid_assets: updated_asset_status = AssetStatus.incomplete break await bulk_session_model.update_bulk_session( bulk_session_id, identity_uuid, UpdateBulkSessionRequest(asset_status=updated_asset_status), ) return updated_asset_status async def get_bulk_session_ids_by_product_ids( product_ids: list[int], identity_uuid: UUID4, product_submit_statuses: tuple[str, ...] | None = None, product_ingestion_statuses: tuple[str] = (IngestionStatus.SUCCESS,), ): """Get bulk session ids using the product ids. Filters 'bulk_session_ingestion_products' by ingestion_status. """ product_and_session_ids = ( await bulk_session_ingestion_product.get_bulk_session_ids_by_product_ids( product_ids, product_submit_statuses, product_ingestion_statuses ) ) product_id_to_sessions = defaultdict(list) for product_and_session_id in product_and_session_ids: product_id_to_sessions[product_and_session_id["product_id"]].append( { "id": product_and_session_id["bulk_session_id"], } ) return [ { "product_id": product_id, "bulk_sessions": product_id_to_sessions.get(product_id, []), } for product_id in product_ids ] async def get_bulk_session_ids_by_track_ids( track_ids: list[int], identity_uuid: UUID4, product_ingestion_statuses=(IngestionStatus.SUCCESS,), ): """Get bulk session ids using the track ids. Filters 'bulk_session_ingestion_products' by ingestion_status. """ track_and_session_ids = ( await bulk_session_ingestion_track.get_bulk_session_ids_by_track_ids( track_ids, product_ingestion_statuses ) ) track_id_to_sessions = defaultdict(list) for track_and_session_id in track_and_session_ids: track_id_to_sessions[track_and_session_id["track_id"]].append( { "id": track_and_session_id["bulk_session_id"], } ) return [ { "track_id": track_id, "bulk_sessions": track_id_to_sessions.get(track_id, []), } for track_id in track_ids ] def _sort_items( items: list, sort_by: SortField, sort_order: SortOrder, get_sort_key_fn ) -> list: """ Generic sorting function for products or projects. """ if not items: return items reverse = sort_order == SortOrder.DESC return sorted( items, key=lambda item: get_sort_key_fn(item, sort_by), reverse=reverse ) def _get_sort_key(item, sort_by: SortField): """ Extract sort key from any item (product or project) based on sort field. """ if sort_by == SortField.NAME: return natural_key(item.name) elif sort_by == SortField.ERRORS: return len(item.errors) else: if hasattr(item, "product_code"): return natural_key(item.product_code) elif hasattr(item, "project_code"): return natural_key(item.project_code) else: return natural_key(item.name) async def assert_access_from_sqs_message( sqs_message: Dict[str, Any], identity_uuid: UUID4, profiles: list[tuple[str, int]] ) -> bool: """The ingest endpoint receives no data so read the id from the SQS message. This function checks if a bulk session is cancelled before processing. If cancelled, the message is deleted from the queue to prevent infinite retries. Returns: bool: True if processing should continue, False if session is cancelled/should skip """ message_attributes = sqs_message["MessageAttributes"] bulk_session_id = message_attributes["BulkSessionId"]["StringValue"] # First check if session exists (including cancelled ones) session = await get_bulk_session( bulk_session_id=bulk_session_id, identity_uuid=identity_uuid, is_cancelled=None, # Check all sessions, not just non-cancelled get_download_link=False, enrich=False, ) # If session is cancelled, delete message and return False to skip processing if session and session.is_cancelled: logger.warning( f"Bulk session {bulk_session_id} is cancelled. " f"Deleting SQS message to prevent reprocessing." ) await delete_ingestion_message_from_sqs(sqs_message) return False # Continue with normal access check for non-cancelled sessions await assert_access( bulk_session_id=bulk_session_id, identity_uuid=identity_uuid, profiles=profiles ) return True @tracer.wrap() async def assert_access( *, bulk_session_id: UUID4, identity_uuid: UUID4, profiles: list[tuple[str, int]] ) -> None: """ Check if the identity can access the bulk session. This will raise AssertionError if the bulk session does not exist or if the vendor UUID is missing. It will raise HTTPException from check_vendor_access_for_profiles if the identity + profiles does not have access to the vendor. Args: bulk_session_id (UUID4): the ID of the bulk session to check identity_uuid (UUID4): the UUID of the identity making the request profiles (list[tuple[str, int]]): profiles specified in the request headers Returns: None Raises: AssertionError: if the bulk session does not exist or if the vendor UUID is missing HTTPException: raised from check_vendor_access_for_profiles """ session = await get_bulk_session( bulk_session_id, identity_uuid, get_download_link=False, enrich=False, ) assert session vendor_uuid = session.vendor_uuid assert vendor_uuid subaccount_id = session.subaccount_id await check_vendor_access_for_profiles( identity_id=identity_uuid, vendor_uuid=UUID(vendor_uuid), profiles=profiles, allowed_types=header.BULK_SESSION_ALLOWED_PROFILE_TYPES, subaccount_id=subaccount_id, ) @tracer.wrap() async def assert_authorization( *, bulk_session_id: UUID4, identity_uuid: UUID4, profiles: list[tuple[str, int]], resource_type: str = "digital_audio", action: str = "bulk_create", ) -> None: """ Check if the identity is authorized to access the bulk session. First checks is_authorized_for_tenant and returns immediately if authorized. If not, falls back to check_vendor_access_for_profiles. Raises AssertionError if the bulk session does not exist or if the vendor UUID is missing. Args: bulk_session_id (UUID4): the ID of the bulk session to check identity_uuid (UUID4): the UUID of the identity making the request profiles (list[tuple[str, int]]): profiles specified in the request headers resource_type (str): the resource type for the authorization check action (str): the action for the authorization check Returns: None Raises: AssertionError: if the bulk session does not exist or if the vendor UUID is missing HTTPException: raised from is_authorized_for_tenant (401) or check_vendor_access_for_profiles (403/404) """ session = await get_bulk_session( bulk_session_id, identity_uuid, get_download_link=False, enrich=False, ) assert session assert session.vendor_uuid vendor_uuid = UUID(session.vendor_uuid) tenant = await get_tenant( vendor_uuid, session.subaccount_id, ) # In future we should be able to get the tenant_type from the session model, passing "account" for now if tenant and is_authorized_for_tenant( tenant_uuid=tenant.tenant_uuid, tenant_type=tenant.tenant_type, tenant_attributes=tenant.tenant_attributes, resource_type=resource_type, action=action, ): return # TODO: when we stop calling this we'll need to add return False or raise a 403 await check_vendor_access_for_profiles( identity_id=identity_uuid, vendor_uuid=vendor_uuid, profiles=profiles, allowed_types=header.BULK_SESSION_ALLOWED_PROFILE_TYPES, subaccount_id=session.subaccount_id, ) @tracer.wrap() async def assert_authorization_many( *, bulk_session_ids: list[UUID4], identity_uuid: UUID4, profiles: list[tuple[str, int]], resource_type: str = "digital_audio", action: str = "bulk_create", ) -> None: """Check if the identity is authorized to access all bulk sessions. First checks is_authorized_for_tenants and returns immediately if all are authorized. If not, falls back to check_vendor_access_for_profiles for only the sessions that ows-pdp denied. Args: bulk_session_ids: the IDs of the bulk sessions to check identity_uuid: the UUID of the identity making the request profiles: profiles specified in the request headers resource_type: the resource type for the authorization check action: the action for the authorization check Raises: AssertionError: if any session is missing vendor_uuid (programmer error) HTTPException: 404 if any bulk session ID is not found; 401 if not authenticated (raised by is_authorized_for_tenants); 500 if tenant and session-id lists are mismatched (programmer error surfaced by is_authorized_for_tenants); 403/404 from the profile access fallback (only for sessions PDP denied). """ bulk_sessions = await get_bulk_sessions(bulk_session_ids) if len(bulk_sessions) != len(bulk_session_ids): raise HTTPException(status_code=404, detail=ERROR_MESSAGE_NO_BULK_SESSION) # Restore caller order — SQL IN(...) doesn't preserve it session_by_id = {bs.bulk_session_id: bs for bs in bulk_sessions} bulk_sessions = [session_by_id[str(bsid)] for bsid in bulk_session_ids] for bulk_session in bulk_sessions: assert bulk_session.vendor_uuid fetched_tenants = await get_tenants_many( [ TenantLookup( vendor_uuid=UUID(bs.vendor_uuid), subaccount_id=bs.subaccount_id, ) for bs in bulk_sessions ] ) # vendor_uuid is asserted above, so get_tenants_many always resolves a Tenant tenants_list: list[Tenant] = cast(list[Tenant], fetched_tenants) session_ids_list: list[UUID4] = [bs.bulk_session_id for bs in bulk_sessions] if tenants_list: pdp_response = is_authorized_for_tenants( tenants=tenants_list, bulk_session_ids=session_ids_list, resource_type=resource_type, action=action, ) if all(pdp_response): return # Perform Profile access checks for bulk sessions that received a DENY from PDP. # TODO: when we stop calling this we'll need to raise a 403 here for bulk_session, authorized in zip(bulk_sessions, pdp_response): if not authorized: await check_vendor_access_for_profiles( identity_id=identity_uuid, vendor_uuid=UUID(bulk_session.vendor_uuid), profiles=profiles, allowed_types=header.BULK_SESSION_ALLOWED_PROFILE_TYPES, subaccount_id=bulk_session.subaccount_id, ) async def get_bulk_session_ids_by_project_ids( project_ids: list[int], identity_uuid: UUID4, project_ingestion_statuses=(IngestionStatus.SUCCESS,), ): """Get bulk session ids using the project ids. Filters 'bulk_session_ingestion_projects' by ingestion_status. """ project_and_session_ids = ( await bulk_session_ingestion_project.get_bulk_session_ids_by_project_ids( project_ids, project_ingestion_statuses ) ) project_id_to_sessions = defaultdict(list) for project_and_session_id in project_and_session_ids: project_id_to_sessions[project_and_session_id["project_id"]].append( { "id": project_and_session_id["bulk_session_id"], } ) return [ { "project_id": project_id, "bulk_sessions": project_id_to_sessions.get(project_id, []), } for project_id in project_ids ]