"""HTTP router for data-access management V1.""" from __future__ import annotations import logging from collections.abc import Mapping from datetime import UTC, datetime from typing import Literal, NoReturn from uuid import UUID, uuid4 from fastapi import APIRouter, Depends, Header, Path, Query, Request, Response from sqlalchemy import ( String, and_, case, cast, distinct, func, literal, or_, select, ) from sqlalchemy.ext.asyncio import AsyncSession from app.approval.models import ApprovalRequest from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( check_approval_submit_rate_limit, get_current_user, ) from app.core.dependencies import ( require_active_employee_for_mutations as _leaver_runtime_gate, ) from app.core.exceptions import AppException from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role from app.core.response import paginated_response, success_response from app.domains.data_access import constants as c from app.domains.data_access.apply_gate import data_access_apply_gate_config from app.domains.data_access.binding_projection import ( binding_access_state, provider_member_for_binding, ) from app.domains.data_access.cutover_report import ( DataAccessCutoverReportItem, build_data_access_cutover_report, ) from app.domains.data_access.lineage import ( DataAccessLineageCatalogError, DataAccessLineageCatalogSearchMatch, DataAccessLineageCatalogTable, get_table_lineage, list_lineage_catalog_tables, search_lineage_catalog_tables, ) from app.domains.data_access.lineage_owners import ( lineage_dataset_owner_fields, resolve_lineage_dataset_owners, ) from app.domains.data_access.models import ( BINDING_STATUS_ACTIVE, BINDING_STATUS_REVOKED, BINDING_STATUS_REVOKED_PENDING, EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_PRESENT, EFFECT_MANAGED_REASON_GRANT_REQUEST, EFFECT_STATUS_APPLIED, EFFECT_STATUS_FAILED, EFFECT_STATUS_PENDING_RECONCILE, EFFECT_STATUS_RETRYABLE, OWNER_RESOLUTION_RESOLVED, OWNER_SOURCE_DATASET_INHERITED, RECONCILE_STATUS_APPLIED, RECONCILE_STATUS_FAILED, RECONCILE_STATUS_NOT_STARTED, RECONCILE_STATUS_OBSERVED, RECONCILE_STATUS_PENDING, RECONCILE_STATUS_RETRYABLE, REQUEST_ITEM_STATUS_APPROVED, REQUEST_ITEM_STATUS_REJECTED, REQUEST_ITEM_STATUS_STALE, REQUEST_STATUS_APPROVED, REQUEST_STATUS_CANCELLED, REQUEST_STATUS_PENDING_FEISHU, REQUEST_STATUS_REJECTED, REQUEST_STATUS_STALE_REQUIRES_RESUBMIT, RESOURCE_STATUS_ACTIVE, RESOURCE_STATUS_DELETED, RESOURCE_STATUS_SUSPENDED, DataAccessBinding, DataAccessCompiledEffect, DataAccessEffectRun, DataAccessEntitlement, DataAccessPrincipal, DataAccessRequest, DataAccessRequestItem, DataAccessResource, DataAccessServiceAccountProfile, ) from app.domains.data_access.my_grants import list_my_data_access_grants from app.domains.data_access.reconciler import ( _refresh_reconcile_rollups, apply_absent_reconcile_compiled_effects, apply_present_reconcile_compiled_effects, ) from app.domains.data_access.schemas import ( DataAccessAdminOverviewApplyModeResponse, DataAccessAdminOverviewBindingsResponse, DataAccessAdminOverviewDatasetsResponse, DataAccessAdminOverviewEffectsResponse, DataAccessAdminOverviewPeopleResponse, DataAccessAdminOverviewRecentRequestResponse, DataAccessAdminOverviewRequestsResponse, DataAccessAdminOverviewResponse, DataAccessAdminOverviewTopDatasetResponse, DataAccessAdminOverviewTopGroupResponse, DataAccessCutoverValidationHistoryResponse, DataAccessCutoverValidationItemResponse, DataAccessCutoverValidationReportResponse, DataAccessCutoverValidationSummaryResponse, DataAccessDatasetFieldSchemaResponse, DataAccessDatasetTableSchemaResponse, DataAccessDatasetTableSummaryResponse, DataAccessFailedApplyItemResponse, DataAccessFailedApplyLatestRunResponse, DataAccessFailedApplyQueueResponse, DataAccessFailedApplyRetryResponse, DataAccessFailedApplySummaryResponse, DataAccessGroupAccessItemResponse, DataAccessGroupAccessLatestRunResponse, DataAccessGroupAccessPrincipalResponse, DataAccessGroupAccessResponse, DataAccessGroupAccessSummaryResponse, DataAccessMyGrantItemResponse, DataAccessOwnerTransferRequest, DataAccessReconcileRunHistoryResponse, DataAccessReconcileRunItemResponse, DataAccessReconcileRunSummaryResponse, DataAccessRequestablePrincipalResponse, DataAccessRequestableResourceResponse, DataAccessRequestableTableMatchResponse, DataAccessRequestDetailApprovalResponse, DataAccessRequestDetailCompiledPreviewResponse, DataAccessRequestDetailEffectResponse, DataAccessRequestDetailItemResponse, DataAccessRequestDetailResponse, DataAccessRequestDetailRunResponse, DataAccessSubmitGrantResponse, DataAccessSubmitPersonalGrantRequest, DataAccessSubmitServiceAccountGrantRequest, DataAccessTableLineageEdgeResponse, DataAccessTableLineageNodeResponse, DataAccessTableLineageResponse, DataAccessTableSelectionRequest, ) from app.domains.data_access.service import ( DATA_ACCESS_COMPILER_VERSION, DataAccessService, ) from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.domains.email_group.service import EmailGroupService from app.identity.person_models import IdentityPerson from app.integrations.driver_base import ( DataAccessDatasetTableSummary, DataAccessDriver, ) from app.integrations.models import Integration from app.integrations.service import build_primary_driver logger = logging.getLogger(__name__) router = APIRouter( prefix="/api/v1/data-access", tags=["data-access"], dependencies=[Depends(_leaver_runtime_gate)], ) DATA_ACCESS_RESOURCE_AVAILABILITY_REQUESTABLE = "requestable" MAX_REQUESTABLE_RESOURCE_TABLE_MATCHES = 20 DATA_ACCESS_RESOURCE_AVAILABILITY_CATALOG_MISSING_SUSPENDED = ( "catalog_missing_suspended" ) DATA_ACCESS_RESOURCE_AVAILABILITY_NOT_REQUESTABLE = "not_requestable" DATA_ACCESS_NOTICE_CATALOG_MISSING_SUSPENSION = "catalog_missing_suspension" DATA_ACCESS_NOTICE_GROUP_DELETED_REVOCATION = "group_deleted_revocation" DATA_ACCESS_REVOKE_REASON_GROUP_DELETED = "group_deleted_without_bridge" async def _requestable_dataset_for_schema_browser( db: AsyncSession, resource_id: UUID, ) -> DataAccessResource: row = await db.scalar( select(DataAccessResource).where( DataAccessResource.id == resource_id, DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessResource.status == RESOURCE_STATUS_ACTIVE, DataAccessResource.requestable.is_(True), DataAccessResource.owner_person_id.is_not(None), DataAccessResource.owner_resolution_status == OWNER_RESOLUTION_RESOLVED, ), ) if row is None: raise AppException( code="DATA_ACCESS_RESOURCE_NOT_REQUESTABLE", message="The requested dataset is not available in the requestable catalog.", status_code=404, ) return row async def _dataset_schema_driver(db: AsyncSession) -> DataAccessDriver: driver = await build_primary_driver(DataAccessDriver.kind, session=db) if not isinstance(driver, DataAccessDriver): raise AppException( code="DATA_ACCESS_DRIVER_INVALID", message="Dataset schema browsing is temporarily unavailable.", status_code=503, ) return driver async def _resolve_submit_resource_ids( db: AsyncSession, *, dataset_resource_ids: list[UUID], table_selections: list[DataAccessTableSelectionRequest], ) -> list[UUID]: resource_ids = list(dataset_resource_ids) if table_selections: resource_ids.extend( await DataAccessService().resolve_table_selection_resource_ids( db, selections=table_selections, driver=await _dataset_schema_driver(db), ), ) return resource_ids def _raise_schema_browser_unavailable( exc: Exception, *, project_id: str, dataset_id: str, table_id: str | None = None, ) -> NoReturn: logger.warning( "DAX schema metadata read failed for %s.%s table=%s error_type=%s", project_id, dataset_id, table_id or "-", type(exc).__name__, ) raise AppException( code="DATA_ACCESS_SCHEMA_UNAVAILABLE", message="Dataset schema metadata is temporarily unavailable.", status_code=503, ) from None @router.get( "/capabilities", summary="Read the data-access V1 domain contract and authority boundary", ) async def get_data_access_capabilities( request: Request, _: AuthUser = Depends(get_current_user), ): service = DataAccessService() data = service.get_capabilities() return success_response( data=data.model_dump(mode="json"), request=request, ) @router.get( "/requestable-resources", summary="List visible BigQuery dataset catalog rows for data-access requests", ) async def list_requestable_resources( request: Request, search: str | None = Query(None, max_length=255), requestability: Literal[ "all", "requestable", "not_requestable" ] = Query("all"), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _require_data_access_request_author(db, current_user) is_global_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE) conditions = [ DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessResource.owner_person_id.is_not(None), DataAccessResource.owner_resolution_status == OWNER_RESOLUTION_RESOLVED, ] active_requestable_condition = and_( DataAccessResource.status == RESOURCE_STATUS_ACTIVE, DataAccessResource.requestable.is_(True), ) if is_global_admin: conditions.append( DataAccessResource.status.in_( [RESOURCE_STATUS_ACTIVE, RESOURCE_STATUS_SUSPENDED], ), ) else: conditions.append(DataAccessResource.status == RESOURCE_STATUS_ACTIVE) if requestability == "requestable": conditions.append(active_requestable_condition) elif requestability == "not_requestable": conditions.append( or_( DataAccessResource.status != RESOURCE_STATUS_ACTIVE, DataAccessResource.requestable.is_(False), ) ) normalized_search = str(search or "").strip() catalog_search_matches: tuple[DataAccessLineageCatalogSearchMatch, ...] = () catalog_matches_by_locator: dict[ tuple[str, str], list[DataAccessLineageCatalogSearchMatch], ] = {} if normalized_search: try: catalog_search_matches = search_lineage_catalog_tables(normalized_search) except DataAccessLineageCatalogError as exc: logger.warning( "DAX unified resource search skipped malformed lineage catalog " "error_type=%s", type(exc).__name__, ) for match in catalog_search_matches: catalog_matches_by_locator.setdefault( (match.project_id, match.dataset_id), [], ).append(match) like = f"%{normalized_search}%" table_resource_match_ids = select( DataAccessResource.parent_resource_id, ).where( DataAccessResource.resource_type == c.RESOURCE_TYPE_TABLE, DataAccessResource.status == RESOURCE_STATUS_ACTIVE, DataAccessResource.parent_resource_id.is_not(None), or_( DataAccessResource.table_id.ilike(like), DataAccessResource.display_name.ilike(like), DataAccessResource.notes.ilike(like), ), ) catalog_dataset_conditions = [ and_( DataAccessResource.project_id == project_id, DataAccessResource.dataset_id == dataset_id, ) for project_id, dataset_id in catalog_matches_by_locator ] conditions.append( or_( DataAccessResource.project_id.ilike(like), DataAccessResource.dataset_id.ilike(like), DataAccessResource.display_name.ilike(like), DataAccessResource.notes.ilike(like), DataAccessResource.id.in_(table_resource_match_ids), *catalog_dataset_conditions, ), ) total = int( await db.scalar( select(func.count()).select_from(DataAccessResource).where(*conditions), ) or 0 ) rows = ( await db.scalars( select(DataAccessResource) .where(*conditions) .order_by( case((active_requestable_condition, 0), else_=1), DataAccessResource.project_id, DataAccessResource.dataset_id, DataAccessResource.id, ) .offset((page - 1) * page_size) .limit(page_size), ) ).all() owner_labels = await _person_labels( db, [UUID(str(row.owner_person_id)) for row in rows if row.owner_person_id], ) managed_matches_by_parent_id: dict[str, list[DataAccessResource]] = {} if normalized_search and rows: like = f"%{normalized_search}%" managed_matches = ( await db.scalars( select(DataAccessResource) .where( DataAccessResource.parent_resource_id.in_([row.id for row in rows]), DataAccessResource.resource_type == c.RESOURCE_TYPE_TABLE, DataAccessResource.status == RESOURCE_STATUS_ACTIVE, or_( DataAccessResource.table_id.ilike(like), DataAccessResource.display_name.ilike(like), DataAccessResource.notes.ilike(like), ), ) .order_by(DataAccessResource.table_id, DataAccessResource.id), ) ).all() for table_resource in managed_matches: managed_matches_by_parent_id.setdefault( str(table_resource.parent_resource_id), [], ).append(table_resource) def matched_tables_for_resource( row: DataAccessResource, ) -> tuple[int, tuple[DataAccessRequestableTableMatchResponse, ...]]: matches_by_table_id: dict[str, DataAccessRequestableTableMatchResponse] = {} for match in catalog_matches_by_locator.get( (row.project_id, row.dataset_id), (), ): matches_by_table_id[match.table.table_id] = ( DataAccessRequestableTableMatchResponse( table_id=match.table.table_id, table_type=match.table.table_type, description=match.table.description, catalog_tier="lineage", ) ) for table_resource in managed_matches_by_parent_id.get(str(row.id), ()): if not table_resource.table_id: continue existing = matches_by_table_id.get(table_resource.table_id) if existing is not None: if existing.description is None and table_resource.notes: matches_by_table_id[table_resource.table_id] = existing.model_copy( update={"description": table_resource.notes}, ) continue matches_by_table_id[table_resource.table_id] = ( DataAccessRequestableTableMatchResponse( table_id=table_resource.table_id, table_type=table_resource.table_type or "TABLE", description=table_resource.notes, catalog_tier="managed", ) ) matches = tuple( matches_by_table_id[table_id] for table_id in sorted(matches_by_table_id, key=str.casefold) ) return len(matches), matches[:MAX_REQUESTABLE_RESOURCE_TABLE_MATCHES] matched_tables_by_resource_id = { str(row.id): matched_tables_for_resource(row) for row in rows } data = [ DataAccessRequestableResourceResponse( id=row.id, provider=row.provider, resource_type=row.resource_type, status=row.status, requestable=row.requestable, availability=_requestable_resource_availability(row), notice_code=_requestable_resource_notice_code(row), project_id=row.project_id, dataset_id=row.dataset_id, display_name=row.display_name, description=row.notes, owner_person_id=row.owner_person_id, owner_label=owner_labels.get(str(row.owner_person_id)), owner_resolution_status=row.owner_resolution_status, owner_authority_enabled=row.resource_owner_authority_enabled, caller_can_manage_owner=( _requestable_resource_availability(row) == "requestable" and str(row.owner_person_id) == str(current_user.id) ), owner_source=row.owner_source, source_version=row.source_version, source_etag=row.source_etag, updated_at=row.updated_at, matched_table_count=matched_tables_by_resource_id[str(row.id)][0], matched_tables=matched_tables_by_resource_id[str(row.id)][1], ).model_dump(mode="json") for row in rows ] return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.put( "/requestable-resources/{resource_id}/owner", summary="Transfer one BigQuery dataset as its current owner", ) async def update_requestable_dataset_owner( request: Request, body: DataAccessOwnerTransferRequest, resource_id: UUID = Path(...), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _require_data_access_request_author(db, current_user) updated = await DataAccessService().transfer_bigquery_dataset_owner( db, dataset_resource_id=resource_id, actor=current_user, new_owner_person_id=body.owner_person_id, trace_id=getattr(request.state, "trace_id", None), ) await db.commit() row = await db.get(DataAccessResource, updated.resource_id) if row is None: # defensive: transfer validates and locks this exact row raise AppException( code="DATA_ACCESS_RESOURCE_NOT_FOUND", message="The transferred dataset is no longer available.", status_code=404, ) owner_labels = await _person_labels(db, [body.owner_person_id]) data = DataAccessRequestableResourceResponse( id=row.id, provider=row.provider, resource_type=row.resource_type, status=row.status, requestable=row.requestable, availability=_requestable_resource_availability(row), notice_code=_requestable_resource_notice_code(row), project_id=row.project_id, dataset_id=row.dataset_id, display_name=row.display_name, description=row.notes, owner_person_id=row.owner_person_id, owner_label=owner_labels.get(str(row.owner_person_id)), owner_resolution_status=row.owner_resolution_status, owner_authority_enabled=row.resource_owner_authority_enabled, caller_can_manage_owner=( str(row.owner_person_id) == str(current_user.id) ), owner_source=row.owner_source, source_version=row.source_version, source_etag=row.source_etag, updated_at=row.updated_at, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/requestable-resources/{resource_id}/tables", summary="List tables in one requestable BigQuery dataset", ) async def list_requestable_dataset_tables( request: Request, resource_id: UUID = Path(...), search: str | None = Query(None, max_length=255), page: int = Query(1, ge=1), page_size: int = Query(100, ge=1, le=200), scope: Literal["all", "tracked", "other"] = Query("all"), cursor: str | None = Query(None, max_length=4096), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _require_data_access_request_author(db, current_user) resource = await _requestable_dataset_for_schema_browser(db, resource_id) table_resource_rows = ( await db.scalars( select(DataAccessResource).where( DataAccessResource.parent_resource_id == resource_id, DataAccessResource.resource_type == c.RESOURCE_TYPE_TABLE, DataAccessResource.status == RESOURCE_STATUS_ACTIVE, ), ) ).all() table_resources = { str(row.table_id): row for row in table_resource_rows if row.table_id } try: lineage_tables = list_lineage_catalog_tables( project_id=resource.project_id, dataset_id=resource.dataset_id, ) except DataAccessLineageCatalogError: lineage_tables = () lineage_by_id = {table.table_id: table for table in lineage_tables} lineage_ids = frozenset(lineage_by_id) managed_ids = frozenset(table_resources) normalized_search = str(search or "").strip().casefold() metadata_source: Literal["provider", "lineage_catalog"] = "provider" provider_available = True next_cursor: str | None = None has_more = False tier_by_id: dict[str, Literal["lineage", "managed", "discovered"]] = {} tables: tuple[ DataAccessDatasetTableSummary | DataAccessLineageCatalogTable, ..., ] page_rows: tuple[ DataAccessDatasetTableSummary | DataAccessLineageCatalogTable, ..., ] if scope == "tracked": if cursor is not None: raise AppException( code="DATA_ACCESS_TABLE_CURSOR_INVALID", message="A table cursor is only valid for the other-table stream.", status_code=422, ) try: await _dataset_schema_driver(db) except Exception as exc: # noqa: BLE001 - integration failures are read-only if isinstance(exc, AppException) and exc.status_code != 503: raise provider_available = False logger.warning( "DAX tracked table list is read-only for %s.%s " "provider_error_type=%s", resource.project_id, resource.dataset_id, type(exc).__name__, ) tracked_by_id: dict[ str, DataAccessDatasetTableSummary | DataAccessLineageCatalogTable, ] = dict(lineage_by_id) for table_id in managed_ids - lineage_ids: tracked_by_id[table_id] = DataAccessDatasetTableSummary( table_id=table_id, table_type="TABLE", ) tables = tuple(sorted( tracked_by_id.values(), key=lambda table: table.table_id.casefold(), )) if normalized_search: tables = tuple( table for table in tables if normalized_search in table.table_id.casefold() ) total = len(tables) start = (page - 1) * page_size page_rows = tables[start:start + page_size] has_more = start + len(page_rows) < total metadata_source = "lineage_catalog" tier_by_id = { table.table_id: ( "lineage" if table.table_id in lineage_ids else "managed" ) for table in page_rows } elif scope == "other": if page_size > 100: raise AppException( code="DATA_ACCESS_TABLE_PAGE_SIZE_INVALID", message="Other-table discovery page_size cannot exceed 100.", status_code=422, ) if cursor is not None and (normalized_search or page != 1): raise AppException( code="DATA_ACCESS_TABLE_CURSOR_INVALID", message="A table cursor cannot be combined with search or page offsets.", status_code=422, ) excluded_ids = lineage_ids | managed_ids try: driver = await _dataset_schema_driver(db) if normalized_search: provider_tables = await driver.list_dataset_tables( resource.project_id, resource.dataset_id, ) tables = tuple( table for table in provider_tables if table.table_id not in excluded_ids and normalized_search in table.table_id.casefold() ) total = len(tables) start = (page - 1) * page_size page_rows = tables[start:start + page_size] has_more = start + len(page_rows) < total else: provider_page = await driver.list_dataset_tables_page( resource.project_id, resource.dataset_id, page_size=page_size, page_token=cursor, ) page_rows = tuple( table for table in provider_page.items if table.table_id not in excluded_ids ) provider_total = provider_page.total_rows total = max( len(page_rows), (provider_total - len(excluded_ids)) if provider_total is not None else len(page_rows), ) next_cursor = provider_page.next_page_token has_more = next_cursor is not None except Exception as exc: # noqa: BLE001 - SDK/provider exceptions vary if isinstance(exc, AppException): raise _raise_schema_browser_unavailable( exc, project_id=resource.project_id, dataset_id=resource.dataset_id, ) tier_by_id = {table.table_id: "discovered" for table in page_rows} else: if cursor is not None: raise AppException( code="DATA_ACCESS_TABLE_CURSOR_INVALID", message="A table cursor is only valid for the other-table stream.", status_code=422, ) try: driver = await _dataset_schema_driver(db) tables = await driver.list_dataset_tables( resource.project_id, resource.dataset_id, ) except Exception as exc: # noqa: BLE001 - SDK/provider exceptions vary if isinstance(exc, AppException) and exc.status_code != 503: raise if not lineage_tables: if isinstance(exc, AppException): raise exc _raise_schema_browser_unavailable( exc, project_id=resource.project_id, dataset_id=resource.dataset_id, ) tables = lineage_tables metadata_source = "lineage_catalog" provider_available = False logger.warning( "DAX table list using read-only lineage catalog for %s.%s " "provider_error_type=%s", resource.project_id, resource.dataset_id, type(exc).__name__, ) if normalized_search: tables = tuple( table for table in tables if normalized_search in table.table_id.casefold() ) total = len(tables) start = (page - 1) * page_size page_rows = tables[start:start + page_size] has_more = start + len(page_rows) < total tier_by_id = { table.table_id: ( "lineage" if table.table_id in lineage_ids else "managed" if table.table_id in managed_ids else "discovered" ) for table in page_rows } effective_owner_resources = { table.table_id: ( table_resources[table.table_id] if table.table_id in table_resources and table_resources[table.table_id].owner_person_id else resource ) for table in page_rows } owner_ids = { UUID(str(owner_resource.owner_person_id)) for owner_resource in effective_owner_resources.values() if owner_resource.owner_person_id } owner_labels = await _person_labels(db, list(owner_ids)) data = [] for table in page_rows: table_resource = table_resources.get(table.table_id) effective_owner = effective_owner_resources[table.table_id] table_has_owner = bool( table_resource is not None and table_resource.owner_person_id, ) effective_owner_source = ( table_resource.owner_source if table_resource is not None and table_resource.owner_person_id else OWNER_SOURCE_DATASET_INHERITED ) data.append( DataAccessDatasetTableSummaryResponse( table_id=table.table_id, table_type=table.table_type, resource_id=( table_resource.id if table_resource is not None else None ), owner_person_id=( str(effective_owner.owner_person_id) if effective_owner.owner_person_id else None ), owner_label=( owner_labels.get(str(effective_owner.owner_person_id)) if effective_owner.owner_person_id else None ), owner_source=effective_owner_source, owner_inherited=( not table_has_owner or effective_owner_source == OWNER_SOURCE_DATASET_INHERITED ), caller_can_manage_owner=( provider_available and str(effective_owner.owner_person_id) == str(current_user.id) ), metadata_source=metadata_source, catalog_tier=tier_by_id[table.table_id], lineage_tracked=table.table_id in lineage_ids, downstream_reference_count=( lineage_by_id[table.table_id].downstream_reference_count if table.table_id in lineage_by_id else 0 ), lineage_direction=( lineage_by_id[table.table_id].lineage_direction if table.table_id in lineage_by_id else None ), selectable=provider_available, schema_available=provider_available, ).model_dump(mode="json"), ) return success_response( data=data, request=request, meta={ "page": page, "page_size": page_size, "total": total, "scope": scope, "next_cursor": next_cursor, "has_more": has_more, }, ) @router.put( "/requestable-resources/{resource_id}/tables/{table_id}/owner", summary="Transfer one BigQuery table as its current owner", ) async def update_requestable_dataset_table_owner( request: Request, body: DataAccessOwnerTransferRequest, resource_id: UUID = Path(...), table_id: str = Path(..., min_length=1, max_length=1024), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _require_data_access_request_author(db, current_user) dataset_resource = await _requestable_dataset_for_schema_browser(db, resource_id) try: driver = await _dataset_schema_driver(db) table_schema = await driver.get_dataset_table_schema( dataset_resource.project_id, dataset_resource.dataset_id, table_id, ) except AppException: raise except Exception as exc: # noqa: BLE001 - provider errors are normalized _raise_schema_browser_unavailable( exc, project_id=dataset_resource.project_id, dataset_id=dataset_resource.dataset_id, table_id=table_id, ) service = DataAccessService() await service.resolve_requestable_bigquery_table_resource( db, dataset_resource_id=resource_id, table_id=table_schema.table_id, table_type=table_schema.table_type, ) updated = await service.reassign_bigquery_table_owner( db, dataset_resource_id=resource_id, table_id=table_schema.table_id, actor=current_user, new_owner_person_id=body.owner_person_id, trace_id=getattr(request.state, "trace_id", None), ) await db.commit() owner_labels = await _person_labels(db, [body.owner_person_id]) data = DataAccessDatasetTableSummaryResponse( table_id=updated.table_id or table_schema.table_id, table_type=updated.table_type or table_schema.table_type, resource_id=updated.resource_id, owner_person_id=updated.owner_person_id, owner_label=owner_labels.get(str(body.owner_person_id)), owner_source=updated.owner_source or OWNER_SOURCE_DATASET_INHERITED, owner_inherited=updated.owner_source == OWNER_SOURCE_DATASET_INHERITED, caller_can_manage_owner=( str(updated.owner_person_id) == str(current_user.id) ), # Keep the reassignment response shape identical to the table-list # endpoint. There is no lineage payload on this narrow mutation path. downstream_reference_count=0, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/requestable-resources/{resource_id}/tables/{table_id}/schema", summary="Read current fields for one table in a requestable BigQuery dataset", ) async def get_requestable_dataset_table_schema( request: Request, resource_id: UUID = Path(...), table_id: str = Path(..., min_length=1, max_length=1024), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _require_data_access_request_author(db, current_user) resource = await _requestable_dataset_for_schema_browser(db, resource_id) try: driver = await _dataset_schema_driver(db) schema = await driver.get_dataset_table_schema( resource.project_id, resource.dataset_id, table_id, ) except AppException: raise except Exception as exc: # noqa: BLE001 - SDK/provider exceptions vary _raise_schema_browser_unavailable( exc, project_id=resource.project_id, dataset_id=resource.dataset_id, table_id=table_id, ) data = DataAccessDatasetTableSchemaResponse( project_id=schema.project_id, dataset_id=schema.dataset_id, table_id=schema.table_id, table_type=schema.table_type, description=schema.description, fields=tuple( DataAccessDatasetFieldSchemaResponse( name=field.name, field_path=field.field_path, field_type=field.field_type, mode=field.mode, description=field.description, depth=field.depth, ordinal_position=field.ordinal_position, ) for field in schema.fields ), ) return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/requestable-resources/{resource_id}/tables/{table_id}/lineage", summary="Read deployed dbt lineage for one requestable table", ) async def get_requestable_dataset_table_lineage( request: Request, resource_id: UUID = Path(...), table_id: str = Path(..., min_length=1, max_length=1024), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): await _require_data_access_request_author(db, current_user) resource = await _requestable_dataset_for_schema_browser(db, resource_id) try: lineage = get_table_lineage( project_id=resource.project_id, dataset_id=resource.dataset_id, table_id=table_id, ) except DataAccessLineageCatalogError as exc: logger.warning( "DAX deployed lineage catalog unavailable for %s.%s.%s: %s", resource.project_id, resource.dataset_id, table_id, type(exc).__name__, ) raise AppException( code="DATA_ACCESS_LINEAGE_UNAVAILABLE", message="Table lineage is temporarily unavailable.", status_code=503, ) from None dataset_owners = await resolve_lineage_dataset_owners( db, integration_id=str(resource.integration_id), nodes=(node for node in lineage.nodes if node.node_type == "source"), ) data = DataAccessTableLineageResponse( status=lineage.status, direction=lineage.direction, catalog_id=lineage.catalog_id, source_version=lineage.source_version, generated_at=lineage.generated_at, project_id=lineage.project_id, dataset_id=lineage.dataset_id, table_id=lineage.table_id, nodes=tuple( DataAccessTableLineageNodeResponse( node_id=node.node_id, project_id=node.project_id, dataset_id=node.dataset_id, table_id=node.table_id, node_type=node.node_type, materialization=node.materialization, description=node.description, layer=node.layer, level=node.level, is_root=node.is_root, is_target=node.is_target, **lineage_dataset_owner_fields(node, dataset_owners), ) for node in lineage.nodes ), edges=tuple( DataAccessTableLineageEdgeResponse( upstream_node_id=edge.upstream_node_id, downstream_node_id=edge.downstream_node_id, relationship=edge.relationship, ) for edge in lineage.edges ), truncated=lineage.truncated, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/requestable-principals", summary="List Google Groups the caller may use for data-access requests", ) async def list_requestable_principals( request: Request, search: str | None = Query(None, max_length=255), include_access_history: bool = Query(False), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): is_global_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE) member_group_ids: set[UUID] = set() if not is_global_admin: member_group_ids = await _data_access_member_group_ids_for_user( db, current_user, ) has_data_access_binding = ( select(DataAccessBinding.id) .where(DataAccessBinding.principal_email_group_id == EmailGroup.id) .exists() ) owner_count_subq = ( select( EmailGroupOwner.email_group_id.label("email_group_id"), func.count(EmailGroupOwner.id).label("owner_count"), ) .group_by(EmailGroupOwner.email_group_id) .subquery() ) owner_exists = ( select(EmailGroupOwner.id) .where( EmailGroupOwner.email_group_id == EmailGroup.id, EmailGroupOwner.person_id == current_user.id, ) .exists() ) conditions = [EmailGroup.google_group_id.is_not(None)] if include_access_history and is_global_admin: conditions.append( or_( EmailGroup.status == "active", has_data_access_binding, ), ) else: conditions.append(EmailGroup.status == "active") if not is_global_admin: if member_group_ids: conditions.append( or_( owner_exists, EmailGroup.id.in_(member_group_ids), ), ) else: conditions.append(owner_exists) if search: like = f"%{search.strip().lower()}%" conditions.append( or_( func.lower(EmailGroup.email).like(like), func.lower(EmailGroup.name).like(like), ), ) count_stmt = select(func.count()).select_from(EmailGroup).where(*conditions) total = int(await db.scalar(count_stmt) or 0) rows = ( await db.execute( select( EmailGroup, func.coalesce(owner_count_subq.c.owner_count, 0).label( "owner_count", ), owner_exists.label("caller_is_owner"), ( EmailGroup.id.in_(member_group_ids) if member_group_ids else literal(False) ).label("caller_is_member"), ) .outerjoin( owner_count_subq, owner_count_subq.c.email_group_id == EmailGroup.id, ) .where(*conditions) .order_by(EmailGroup.email, EmailGroup.id) .offset((page - 1) * page_size) .limit(page_size), ) ).all() data = [ DataAccessRequestablePrincipalResponse( id=group.id, email=group.email, name=group.name, description=group.description, status=group.status, google_group_id=str(group.google_group_id), aliases=tuple(sorted(group.aliases or [])), direct_members_count=group.direct_members_count, owner_count=int(owner_count or 0), caller_is_owner=bool(caller_is_owner), caller_is_member=bool(caller_is_member), ).model_dump(mode="json") for group, owner_count, caller_is_owner, caller_is_member in rows ] return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.get( "/principals/{principal_email_group_id}/access", summary="List current and historical BigQuery dataset access for one Google Group", ) async def get_data_access_group_access( request: Request, principal_email_group_id: UUID = Path(...), search: str | None = Query(None, max_length=255), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): principal, caller_is_owner, owner_count = ( await _resolve_data_access_group_access_reader( db, current_user, principal_email_group_id, ) ) conditions = [ DataAccessBinding.principal_email_group_id == principal.id, DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEntitlement.domain == c.DOMAIN_DATA_ACCESS, DataAccessEntitlement.provider == c.PROVIDER_BIGQUERY, DataAccessEntitlement.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEntitlement.principal_type == c.PRINCIPAL_TYPE_GOOGLE_GROUP, DataAccessEntitlement.entitlement == c.ENTITLEMENT_READ, ] if search: like = f"%{search.strip()}%" conditions.append( or_( DataAccessResource.project_id.ilike(like), DataAccessResource.dataset_id.ilike(like), DataAccessResource.display_name.ilike(like), ), ) binding_rows = ( await db.execute( select(DataAccessBinding, DataAccessResource, DataAccessEntitlement) .join( DataAccessResource, DataAccessResource.id == DataAccessBinding.resource_id, ) .join( DataAccessEntitlement, DataAccessEntitlement.id == DataAccessBinding.entitlement_id, ) .where(*conditions) .order_by( DataAccessBinding.status, DataAccessResource.project_id, DataAccessResource.dataset_id, DataAccessBinding.created_at.desc(), ), ) ).all() bindings = [binding for binding, _, _ in binding_rows] binding_ids = [binding.id for binding in bindings] effects_by_binding: dict[UUID, DataAccessCompiledEffect] = {} if binding_ids: effects = ( await db.scalars( select(DataAccessCompiledEffect) .where(DataAccessCompiledEffect.binding_id.in_(binding_ids)) .order_by( DataAccessCompiledEffect.created_at.desc(), DataAccessCompiledEffect.id.desc(), ), ) ).all() for effect in effects: effects_by_binding.setdefault(effect.binding_id, effect) runs_by_effect: dict[UUID, DataAccessEffectRun] = {} effect_ids = [effect.id for effect in effects_by_binding.values()] if effect_ids: runs = ( await db.scalars( select(DataAccessEffectRun) .where(DataAccessEffectRun.effect_id.in_(effect_ids)) .order_by( DataAccessEffectRun.completed_at.desc(), DataAccessEffectRun.id.desc(), ), ) ).all() for run in runs: runs_by_effect.setdefault(run.effect_id, run) owner_labels = await _person_labels( db, [ UUID(str(resource.owner_person_id)) for _, resource, _ in binding_rows if resource.owner_person_id ], ) items = [ _group_access_item_response( binding=binding, resource=resource, entitlement=entitlement, owner_label=owner_labels.get(str(resource.owner_person_id)), effect=effects_by_binding.get(binding.id), latest_run=( runs_by_effect.get(effects_by_binding[binding.id].id) if binding.id in effects_by_binding else None ), ) for binding, resource, entitlement in binding_rows ] items.sort(key=_group_access_item_sort_key) data = DataAccessGroupAccessResponse( principal=DataAccessGroupAccessPrincipalResponse( id=principal.id, email=principal.email, name=principal.name, status=principal.status, google_group_id=str(principal.google_group_id), direct_members_count=principal.direct_members_count, owner_count=owner_count, caller_is_owner=caller_is_owner, ), summary=_group_access_summary(items), items=items, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/me/grants", summary="List BigQuery dataset grants created by the authenticated employee", ) async def list_my_data_access_grants_endpoint( request: Request, search: str | None = Query(None, max_length=255), principal_type: Literal[ "google_user", "google_group", "gcp_service_account", ] | None = Query(None), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=200), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await list_my_data_access_grants( db, requester_person_id=current_user.id, search=search, principal_type=principal_type, page=page, page_size=page_size, ) owner_labels = await _person_labels( db, [ UUID(str(item.resource.owner_person_id)) for item in result.items if item.resource.owner_person_id ], ) from app.domains.data_access.lifecycle import ( # noqa: PLC0415 lifecycle_fields, load_lifecycle_settings, ) lifecycle_settings = await load_lifecycle_settings(db) now = datetime.now(UTC) sa_principal_ids = [ item.binding.principal_id for item in result.items if item.binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT and item.binding.principal_id is not None ] owned_principals: set[UUID] = set() if sa_principal_ids: owned_principals = set((await db.scalars( select(DataAccessServiceAccountProfile.principal_id).where( DataAccessServiceAccountProfile.principal_id.in_(sa_principal_ids), DataAccessServiceAccountProfile.technical_owner_person_id == current_user.id, ), )).all()) data = [] for item in result.items: binding = item.binding effect = item.effect lifecycle = lifecycle_fields(binding, now=now, settings=lifecycle_settings) data.append(DataAccessMyGrantItemResponse( **lifecycle, revoke_reason=binding.revoke_reason, owner_confirmation_actionable=bool( binding.principal_id is not None and binding.principal_id in owned_principals and lifecycle["owner_confirmation_due_at"] is not None ), binding_id=binding.id, resource_id=item.resource.id, source_request_id=item.request.id, source_request_submitted_at=item.request.submitted_at, project_id=item.resource.project_id, dataset_id=item.resource.dataset_id, table_id=item.resource.table_id, resource_type=item.resource.resource_type, locator=".".join(filter(None, ( item.resource.project_id, item.resource.dataset_id, item.resource.table_id, ))), display_name=item.resource.display_name, description=item.resource.notes, owner_label=owner_labels.get(str(item.resource.owner_person_id)), principal_type=binding.principal_type_snapshot, principal_email=binding.principal_email_snapshot, provider_member=( effect.provider_member if effect else provider_member_for_binding(binding) ), provider_role=( effect.provider_role if effect else item.entitlement.provider_role ), entitlement=binding.entitlement_snapshot, access_state=binding_access_state(binding, effect), binding_status=binding.status, reconcile_status=binding.reconcile_status, latest_provider_outcome=( effect.last_provider_outcome if effect else None ), latest_error=( effect.last_error if effect and effect.last_error else binding.last_reconcile_error ), granted_at=binding.granted_at, revoked_at=binding.revoked_at, last_reconcile_at=binding.last_reconcile_at, ).model_dump(mode="json")) return paginated_response( data=data, total=result.total, page=result.page, page_size=result.page_size, request=request, ) @router.get( "/admin/failed-apply", summary="List failed or retryable data-access apply effects for admins", ) async def list_data_access_failed_apply_queue( request: Request, search: str | None = Query(None, max_length=255), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_data_access_global_admin(current_user) conditions = _failed_apply_conditions(search=search, retryable_only=True) total = int( await db.scalar( select(func.count()) .select_from(DataAccessCompiledEffect) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .join( DataAccessResource, DataAccessResource.id == DataAccessCompiledEffect.resource_id, ) .join( DataAccessEntitlement, DataAccessEntitlement.id == DataAccessCompiledEffect.entitlement_id, ) .join( EmailGroup, EmailGroup.id == DataAccessCompiledEffect.principal_email_group_id, ) .where(*conditions), ) or 0 ) summary = await _failed_apply_summary(db, conditions=conditions, total=total) rows = ( await db.execute( _failed_apply_row_select() .where(*conditions) .order_by( DataAccessCompiledEffect.updated_at.desc(), DataAccessCompiledEffect.created_at.desc(), DataAccessCompiledEffect.id, ) .offset((page - 1) * page_size) .limit(page_size), ) ).all() latest_runs = await _latest_failed_apply_runs( db, [effect.id for effect, _, _, _, _ in rows], ) data = DataAccessFailedApplyQueueResponse( summary=summary, items=[ _failed_apply_item_response( effect=effect, binding=binding, resource=resource, entitlement=entitlement, principal=principal, latest_run=latest_runs.get(effect.id), ) for effect, binding, resource, entitlement, principal in rows ], page=page, page_size=page_size, total=total, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/admin/reconcile-runs", summary="List data-access reconciler run history for admins", ) async def list_data_access_reconcile_runs( request: Request, search: str | None = Query(None, max_length=255), mode: str | None = Query(None, max_length=32), effect_status: str | None = Query(None, max_length=32), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_data_access_global_admin(current_user) conditions = _reconcile_run_conditions( search=search, mode=mode, effect_status=effect_status, ) total = int( await db.scalar( select(func.count()) .select_from(DataAccessEffectRun) .join( DataAccessCompiledEffect, DataAccessCompiledEffect.id == DataAccessEffectRun.effect_id, ) .join( DataAccessBinding, DataAccessBinding.id == DataAccessEffectRun.binding_id, ) .join( DataAccessResource, DataAccessResource.id == DataAccessEffectRun.resource_id, ) .join( DataAccessEntitlement, DataAccessEntitlement.id == DataAccessCompiledEffect.entitlement_id, ) .join( EmailGroup, EmailGroup.id == DataAccessCompiledEffect.principal_email_group_id, ) .where(*conditions), ) or 0 ) summary = await _reconcile_run_summary(db, conditions=conditions, total=total) rows = ( await db.execute( _reconcile_run_row_select() .where(*conditions) .order_by( DataAccessEffectRun.completed_at.desc(), DataAccessEffectRun.created_at.desc(), DataAccessEffectRun.id.desc(), ) .offset((page - 1) * page_size) .limit(page_size), ) ).all() data = DataAccessReconcileRunHistoryResponse( summary=summary, items=[ _reconcile_run_item_response( run=run, effect=effect, binding=binding, resource=resource, principal=principal, ) for run, effect, binding, resource, principal in rows ], page=page, page_size=page_size, total=total, ) return success_response(data=data.model_dump(mode="json"), request=request) def _count_case(condition) -> object: return func.coalesce(func.sum(case((condition, 1), else_=0)), 0) async def _build_data_access_admin_overview( db: AsyncSession, ) -> DataAccessAdminOverviewResponse: # ---- Datasets (BigQuery dataset catalog) ------------------------------- ds = ( await db.execute( select( func.count(), _count_case(DataAccessResource.status == RESOURCE_STATUS_ACTIVE), _count_case( and_( DataAccessResource.status == RESOURCE_STATUS_ACTIVE, DataAccessResource.requestable.is_(True), ), ), _count_case(DataAccessResource.status == RESOURCE_STATUS_SUSPENDED), _count_case(DataAccessResource.status == RESOURCE_STATUS_DELETED), ).where( DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type == c.RESOURCE_TYPE_DATASET, ), ) ).one() datasets = DataAccessAdminOverviewDatasetsResponse( total=int(ds[0] or 0), active=int(ds[1] or 0), requestable=int(ds[2] or 0), suspended=int(ds[3] or 0), deleted=int(ds[4] or 0), ) # ---- Bindings (group→dataset grants = "permissions") ------------------- b = ( await db.execute( select( _count_case(DataAccessBinding.status == BINDING_STATUS_ACTIVE), _count_case( DataAccessBinding.status == BINDING_STATUS_REVOKED_PENDING, ), _count_case(DataAccessBinding.status == BINDING_STATUS_REVOKED), func.count( distinct( case( ( DataAccessBinding.status == BINDING_STATUS_ACTIVE, DataAccessBinding.principal_email_group_id, ), ), ), ), func.count( distinct( case( ( DataAccessBinding.status == BINDING_STATUS_ACTIVE, DataAccessBinding.resource_id, ), ), ), ), ), ) ).one() bindings = DataAccessAdminOverviewBindingsResponse( active=int(b[0] or 0), revoked_pending=int(b[1] or 0), revoked_history=int(b[2] or 0), groups_with_access=int(b[3] or 0), datasets_with_access=int(b[4] or 0), ) # ---- People reach (members of groups that hold active access) ---------- active_group_ids = ( select(DataAccessBinding.principal_email_group_id) .where(DataAccessBinding.status == BINDING_STATUS_ACTIVE) .distinct() ) p = ( await db.execute( select( func.coalesce(func.sum(EmailGroup.direct_members_count), 0), _count_case(EmailGroup.direct_members_count.is_(None)), ).where(EmailGroup.id.in_(active_group_ids)), ) ).one() people = DataAccessAdminOverviewPeopleResponse( reach=int(p[0] or 0), groups_with_unknown_size=int(p[1] or 0), ) # ---- Requests ---------------------------------------------------------- in_progress_reconcile = ( RECONCILE_STATUS_NOT_STARTED, RECONCILE_STATUS_PENDING, RECONCILE_STATUS_OBSERVED, RECONCILE_STATUS_RETRYABLE, ) r = ( await db.execute( select( func.count(), _count_case( DataAccessRequest.status == REQUEST_STATUS_PENDING_FEISHU, ), _count_case(DataAccessRequest.status == REQUEST_STATUS_APPROVED), _count_case(DataAccessRequest.status == REQUEST_STATUS_REJECTED), _count_case( DataAccessRequest.status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT, ), _count_case(DataAccessRequest.status == REQUEST_STATUS_CANCELLED), _count_case( DataAccessRequest.reconcile_status == RECONCILE_STATUS_APPLIED, ), _count_case( DataAccessRequest.reconcile_status.in_(in_progress_reconcile), ), _count_case( DataAccessRequest.reconcile_status == RECONCILE_STATUS_FAILED, ), ), ) ).one() requests = DataAccessAdminOverviewRequestsResponse( total=int(r[0] or 0), pending_approval=int(r[1] or 0), approved=int(r[2] or 0), rejected=int(r[3] or 0), stale=int(r[4] or 0), cancelled=int(r[5] or 0), reconcile_applied=int(r[6] or 0), reconcile_in_progress=int(r[7] or 0), reconcile_failed=int(r[8] or 0), ) # ---- Effects health ---------------------------------------------------- e = ( await db.execute( select( _count_case( DataAccessCompiledEffect.status == EFFECT_STATUS_FAILED, ), _count_case( DataAccessCompiledEffect.status == EFFECT_STATUS_RETRYABLE, ), _count_case( DataAccessCompiledEffect.status == EFFECT_STATUS_APPLIED, ), _count_case( DataAccessCompiledEffect.status == EFFECT_STATUS_PENDING_RECONCILE, ), ).where( DataAccessCompiledEffect.provider == c.PROVIDER_BIGQUERY, DataAccessCompiledEffect.resource_type == c.RESOURCE_TYPE_DATASET, ), ) ).one() effects = DataAccessAdminOverviewEffectsResponse( failed=int(e[0] or 0), retryable=int(e[1] or 0), applied=int(e[2] or 0), pending_reconcile=int(e[3] or 0), ) # ---- Apply mode (does the reconciler write to BigQuery yet?) ----------- integrations = ( await db.scalars( select(Integration).where( Integration.kind == c.DOMAIN_DATA_ACCESS, Integration.provider == c.PROVIDER_BIGQUERY, Integration.enabled.is_(True), ), ) ).all() apply_enabled_count = 0 for integration in integrations: config_public = integration.config_public if config_public is not None and not isinstance(config_public, Mapping): continue try: if data_access_apply_gate_config(config_public).apply_mode_enabled: apply_enabled_count += 1 except ValueError: continue apply_mode = DataAccessAdminOverviewApplyModeResponse( integration_count=len(integrations), apply_enabled_count=apply_enabled_count, provider_mutation_enabled=apply_enabled_count > 0, ) # ---- Top datasets by active group count -------------------------------- top_dataset_rows = ( await db.execute( select( DataAccessResource.id, DataAccessResource.project_id, DataAccessResource.dataset_id, DataAccessResource.display_name, func.count().label("cnt"), ) .join( DataAccessBinding, DataAccessBinding.resource_id == DataAccessResource.id, ) .where(DataAccessBinding.status == BINDING_STATUS_ACTIVE) .group_by( DataAccessResource.id, DataAccessResource.project_id, DataAccessResource.dataset_id, DataAccessResource.display_name, ) .order_by( func.count().desc(), DataAccessResource.project_id, DataAccessResource.dataset_id, ) .limit(8), ) ).all() top_datasets = [ DataAccessAdminOverviewTopDatasetResponse( resource_id=row.id, project_id=row.project_id, dataset_id=row.dataset_id, locator=f"{row.project_id}.{row.dataset_id}", display_name=row.display_name, active_group_count=int(row.cnt or 0), ) for row in top_dataset_rows ] # ---- Top groups by active dataset count -------------------------------- top_group_rows = ( await db.execute( select( EmailGroup.id, EmailGroup.email, EmailGroup.name, EmailGroup.direct_members_count, func.count().label("cnt"), ) .join( DataAccessBinding, DataAccessBinding.principal_email_group_id == EmailGroup.id, ) .where(DataAccessBinding.status == BINDING_STATUS_ACTIVE) .group_by( EmailGroup.id, EmailGroup.email, EmailGroup.name, EmailGroup.direct_members_count, ) .order_by(func.count().desc(), EmailGroup.email) .limit(8), ) ).all() top_groups = [ DataAccessAdminOverviewTopGroupResponse( principal_email_group_id=row.id, email=row.email, name=row.name, direct_members_count=row.direct_members_count, active_dataset_count=int(row.cnt or 0), ) for row in top_group_rows ] # ---- Recent requests --------------------------------------------------- item_count_sq = ( select(func.count()) .where(DataAccessRequestItem.request_id == DataAccessRequest.id) .correlate(DataAccessRequest) .scalar_subquery() ) recent_rows = ( await db.execute( select( DataAccessRequest.id, DataAccessRequest.requester_person_id, DataAccessRequest.principal_email_snapshot, DataAccessRequest.status, DataAccessRequest.reconcile_status, DataAccessRequest.submitted_at, item_count_sq.label("dataset_count"), ) .order_by( DataAccessRequest.submitted_at.desc(), DataAccessRequest.id.desc(), ) .limit(8), ) ).all() labels = await _person_labels( db, [row.requester_person_id for row in recent_rows], ) recent_requests = [ DataAccessAdminOverviewRecentRequestResponse( id=row.id, requester_label=labels.get(str(row.requester_person_id)), principal_email=row.principal_email_snapshot, dataset_count=int(row.dataset_count or 0), status=row.status, reconcile_status=row.reconcile_status, submitted_at=row.submitted_at, ) for row in recent_rows ] return DataAccessAdminOverviewResponse( generated_at=datetime.now(UTC), datasets=datasets, bindings=bindings, people=people, requests=requests, effects=effects, apply_mode=apply_mode, top_datasets=top_datasets, top_groups=top_groups, recent_requests=recent_requests, ) @router.get( "/admin/overview", summary="Aggregated data-access governance overview for admins", ) async def get_data_access_admin_overview( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_data_access_global_admin(current_user) data = await _build_data_access_admin_overview(db) return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/admin/cutover-validation", summary="Report data-access catalog cutover readiness for admins", ) async def get_data_access_cutover_validation_report( request: Request, integration_id: str | None = Query(None, max_length=36), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_data_access_global_admin(current_user) report = await build_data_access_cutover_report( db, integration_id=integration_id, ) total = len(report.items) start = (page - 1) * page_size page_items = report.items[start : start + page_size] data = DataAccessCutoverValidationReportResponse( integration_id=report.integration_id, integration_count=report.integration_count, summary=DataAccessCutoverValidationSummaryResponse( total_resources=report.summary.total_resources, cutover_ready=report.summary.cutover_ready, blocking_count=report.summary.blocking_count, category_counts=report.summary.category_counts, blocking_category_counts=report.summary.blocking_category_counts, provider_mutation_attempted=( report.summary.provider_mutation_attempted ), request_handlers_mutate_provider=( report.summary.request_handlers_mutate_provider ), feishu_callbacks_mutate_provider=( report.summary.feishu_callbacks_mutate_provider ), ), items=[_cutover_validation_item_response(item) for item in page_items], page=page, page_size=page_size, total=total, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.post( "/admin/failed-apply/{effect_id}/retry", summary="Retry one failed data-access apply effect through the reconciler", ) async def retry_data_access_failed_apply_effect( request: Request, effect_id: UUID = Path(...), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_data_access_global_admin(current_user) row = await _load_failed_apply_row_for_retry(db, effect_id) if row is None: raise AppException( code="DATA_ACCESS_FAILED_APPLY_NOT_FOUND", message="Retryable data-access failed apply effect not found.", status_code=404, ) effect, binding, resource, entitlement, principal = row previous_status = effect.status retry_run_id = uuid4() retry_trace_id = str( getattr(request.state, "trace_id", None) or f"data_access_admin_retry:{effect.id}" )[:128] now = datetime.now(UTC) effect.status = EFFECT_STATUS_RETRYABLE _record_failed_apply_retry_requested( db, effect=effect, binding=binding, resource=resource, retry_run_id=retry_run_id, trace_id=retry_trace_id, started_at=now, ) await _refresh_reconcile_rollups( db, effects=[effect], run_id=retry_run_id, observed_at=now, ) await AuditService(db).log( trace_id=retry_trace_id, actor_type="user", action="data_access.failed_apply.retry", status="success", actor_user_id=current_user.id, resource_type="data_access_compiled_effect", resource_id=str(effect.id), detail={ "effect_id": str(effect.id), "binding_id": str(binding.id), "resource_id": str(resource.id), "integration_id": str(resource.integration_id), "previous_status": previous_status, "status": effect.status, "desired_state": effect.desired_state, "provider_mutation_owner": "data_access_reconciler", "approval_request_created": False, }, ip_address=request.client.host if request.client else None, user_agent=request.headers.get("user-agent"), ) await db.flush() if effect.desired_state == EFFECT_DESIRED_ABSENT: apply_result = await apply_absent_reconcile_compiled_effects( db, integration_id=str(resource.integration_id), effect_ids=[effect.id], trace_id=retry_trace_id, ) apply_mode = "apply_absent" else: apply_result = await apply_present_reconcile_compiled_effects( db, integration_id=str(resource.integration_id), effect_ids=[effect.id], trace_id=retry_trace_id, ) apply_mode = "apply_present" await db.flush() latest_run = await _latest_failed_apply_run(db, effect.id) item = _failed_apply_item_response( effect=effect, binding=binding, resource=resource, entitlement=entitlement, principal=principal, latest_run=latest_run, ) data = DataAccessFailedApplyRetryResponse( effect_id=effect.id, previous_status=previous_status, status=effect.status, desired_state=effect.desired_state, retry_run_id=retry_run_id, apply_run_id=apply_result.run_id, apply_mode=apply_mode, provider_write_count=apply_result.provider_write_count, result_count=len(apply_result.results), blocked_effect_count=apply_result.blocked_effect_count, config_error=apply_result.config_error, item=item, ) await db.commit() return success_response(data=data.model_dump(mode="json"), request=request) @router.get( "/requests/{request_id}", summary="Read one data-access request with approval, item, and apply status", ) async def get_data_access_request_detail( request: Request, request_id: UUID = Path(...), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): request_row = await db.get(DataAccessRequest, request_id) if request_row is None: _raise_request_not_found() detail_scope = await _resolve_data_access_request_detail_reader_scope( db, current_user, request_row, ) item_rows = ( await db.execute( select(DataAccessRequestItem, DataAccessResource) .join( DataAccessResource, DataAccessResource.id == DataAccessRequestItem.resource_id, ) .where(DataAccessRequestItem.request_id == request_row.id) .order_by(DataAccessRequestItem.line_no), ) ).all() if detail_scope.visible_resource_ids is not None: item_rows = [ (item, resource) for item, resource in item_rows if resource.id in detail_scope.visible_resource_ids ] if not item_rows: _raise_request_not_found() visible_item_ids = {item.id for item, _ in item_rows} bindings = ( await db.scalars( _scope_request_bindings_query( request_id=request_row.id, visible_item_ids=visible_item_ids, detail_scope=detail_scope, ) .where(DataAccessBinding.source_request_id == request_row.id) .order_by(DataAccessBinding.created_at.desc(), DataAccessBinding.id), ) ).all() binding_by_item: dict[UUID, DataAccessBinding] = {} for binding in bindings: if binding.source_request_item_id is None: continue binding_by_item.setdefault(binding.source_request_item_id, binding) effects = ( await db.scalars( _scope_request_effects_query( visible_item_ids=visible_item_ids, detail_scope=detail_scope, ) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .where(DataAccessBinding.source_request_id == request_row.id) .order_by( DataAccessCompiledEffect.created_at.desc(), DataAccessCompiledEffect.id, ), ) ).all() effect_by_binding: dict[UUID, DataAccessCompiledEffect] = {} for effect in effects: effect_by_binding.setdefault(effect.binding_id, effect) visible_effect_ids = {effect.id for effect in effects} runs = ( await db.scalars( _scope_request_effect_runs_query( request_id=request_row.id, visible_effect_ids=visible_effect_ids, detail_scope=detail_scope, ) .order_by(DataAccessEffectRun.completed_at.desc(), DataAccessEffectRun.id) .limit(200), ) ).all() runs_by_effect: dict[UUID, list[DataAccessEffectRun]] = {} for run in runs: runs_by_effect.setdefault(run.effect_id, []).append(run) entitlement = await db.get(DataAccessEntitlement, request_row.entitlement_id) principal = ( await db.get(EmailGroup, request_row.principal_email_group_id) if request_row.principal_email_group_id else None ) data_access_principal = ( await db.get(DataAccessPrincipal, request_row.principal_id) if request_row.principal_id else None ) approval = ( await db.get(ApprovalRequest, request_row.approval_request_id) if request_row.approval_request_id else None ) owner_ids = [ UUID(str(row.owner_person_id)) for _, row in item_rows if row.owner_person_id ] owner_ids.append(UUID(str(request_row.requester_person_id))) owner_labels = await _person_labels(db, owner_ids) items = [ _detail_item_response( item=item, resource=resource, request_row=request_row, entitlement=entitlement, owner_label=owner_labels.get(str(resource.owner_person_id)), binding=binding_by_item.get(item.id), effect=( effect_by_binding.get(binding_by_item[item.id].id) if item.id in binding_by_item else None ), runs_by_effect=runs_by_effect, ) for item, resource in item_rows ] reconcile_status = ( request_row.reconcile_status if detail_scope.full_request else _visible_items_reconcile_status(items) ) last_reconcile_run_id = ( request_row.last_reconcile_run_id if detail_scope.full_request else _visible_items_last_run_id(items) ) last_reconcile_at = ( request_row.last_reconcile_at if detail_scope.full_request else _visible_items_last_reconcile_at(items) ) last_reconcile_error = ( request_row.last_reconcile_error if detail_scope.full_request else _visible_items_last_error(items) ) data = DataAccessRequestDetailResponse( id=request_row.id, requester_person_id=request_row.requester_person_id, requester_label=owner_labels.get(str(request_row.requester_person_id)), principal_email_group_id=request_row.principal_email_group_id, principal_id=request_row.principal_id, principal_type=request_row.principal_type_snapshot, principal_email_snapshot=request_row.principal_email_snapshot, principal_google_group_id_snapshot=( request_row.principal_google_group_id_snapshot ), principal_name=( principal.name if principal else ( data_access_principal.email_snapshot if data_access_principal else None ) ), principal_status=( principal.status if principal else ( data_access_principal.status if data_access_principal else None ) ), entitlement=request_row.entitlement_snapshot, entitlement_version=request_row.entitlement_version_snapshot, provider_role=( entitlement.provider_role if entitlement else c.BIGQUERY_DATA_VIEWER_ROLE ), status=request_row.status, reconcile_status=reconcile_status, feishu_instance_code=request_row.feishu_instance_code, approval_request_id=request_row.approval_request_id, approval=( DataAccessRequestDetailApprovalResponse( id=approval.id, external_id=approval.external_id, approval_type=approval.approval_type, status=approval.status, updated_at=approval.updated_at, ) if approval else None ), business_reason=( request_row.business_reason if detail_scope.full_request else None ), trace_id=request_row.trace_id if detail_scope.full_request else None, submitted_at=request_row.submitted_at, decided_at=request_row.decided_at, last_reconcile_run_id=last_reconcile_run_id, last_reconcile_at=last_reconcile_at, last_reconcile_error=last_reconcile_error, item_count=len(items), items=items, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.post( "/requests/service-account-grants", summary="Submit a Feishu-backed BigQuery resource read request for one GCP service account", dependencies=[Depends(check_approval_submit_rate_limit)], ) async def submit_service_account_grant_request( body: DataAccessSubmitServiceAccountGrantRequest, request: Request, response: Response, current_user: AuthUser = Depends(get_current_user), idempotency_key: str = Header(..., alias="Idempotency-Key", max_length=128), db: AsyncSession = Depends(get_db), ): resource_ids = await _resolve_submit_resource_ids( db, dataset_resource_ids=body.resource_ids, table_selections=body.table_selections, ) detail_url = body.detail_url or ( f"{str(request.base_url).rstrip('/')}/my/data-access/requests" ) result = await DataAccessService().submit_service_account_grant_request( db, requester=current_user, service_account_email=body.service_account_email, resource_ids=resource_ids, business_reason=body.business_reason, detail_url=detail_url, idempotency_key=idempotency_key, trace_id=getattr(request.state, "trace_id", None), ) await db.commit() response.status_code = 200 if result.reused_existing else 201 response.headers["Location"] = f"/api/v1/data-access/requests/{result.request_id}" data = DataAccessSubmitGrantResponse( request_id=result.request_id, approval_request_id=result.approval_request_id, feishu_instance_code=result.feishu_instance_code, feishu_instance_created=not result.reused_existing, reused_existing=result.reused_existing, status=result.status, dataset_count=result.dataset_count, ) return success_response(data=data.model_dump(mode="json"), request=request) @router.post( "/requests/personal-grants", summary="Submit a Feishu-backed BigQuery resource read request for the authenticated employee", dependencies=[Depends(check_approval_submit_rate_limit)], ) async def submit_personal_grant_request( body: DataAccessSubmitPersonalGrantRequest, request: Request, response: Response, current_user: AuthUser = Depends(get_current_user), idempotency_key: str = Header(..., alias="Idempotency-Key", max_length=128), db: AsyncSession = Depends(get_db), ): resource_ids = await _resolve_submit_resource_ids( db, dataset_resource_ids=body.resource_ids, table_selections=body.table_selections, ) detail_url = body.detail_url or ( f"{str(request.base_url).rstrip('/')}/my/data-access/requests" ) result = await DataAccessService().submit_personal_grant_request( db, requester=current_user, resource_ids=resource_ids, business_reason=body.business_reason, detail_url=detail_url, idempotency_key=idempotency_key, trace_id=getattr(request.state, "trace_id", None), ) await db.commit() response.status_code = 200 if result.reused_existing else 201 response.headers["Location"] = f"/api/v1/data-access/requests/{result.request_id}" data = DataAccessSubmitGrantResponse( request_id=result.request_id, approval_request_id=result.approval_request_id, feishu_instance_code=result.feishu_instance_code, feishu_instance_created=not result.reused_existing, reused_existing=result.reused_existing, status=result.status, dataset_count=result.dataset_count, ) return success_response(data=data.model_dump(mode="json"), request=request) async def _person_labels( db: AsyncSession, person_ids: list[UUID], ) -> dict[str, str]: if not person_ids: return {} rows = ( await db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ).where(IdentityPerson.id.in_(person_ids)), ) ).all() labels: dict[str, str] = {} for person_id, display_name, work_email in rows: labels[str(person_id)] = display_name or work_email or str(person_id) return labels def _require_data_access_global_admin(current_user: AuthUser) -> None: if has_role(current_user.role_names, GLOBAL_ADMIN_ROLE): return raise AppException( code="DATA_ACCESS_ADMIN_REQUIRED", message="Only global admins may operate data-access failed applies.", status_code=403, ) def _failed_apply_conditions( *, search: str | None, retryable_only: bool, ): conditions = [ DataAccessCompiledEffect.provider == c.PROVIDER_BIGQUERY, DataAccessCompiledEffect.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessCompiledEffect.provider_member.like( f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}%", ), DataAccessCompiledEffect.provider_role == c.BIGQUERY_DATA_VIEWER_ROLE, DataAccessCompiledEffect.desired_state.in_( (EFFECT_DESIRED_PRESENT, EFFECT_DESIRED_ABSENT), ), DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEntitlement.domain == c.DOMAIN_DATA_ACCESS, DataAccessEntitlement.provider == c.PROVIDER_BIGQUERY, DataAccessEntitlement.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEntitlement.principal_type == c.PRINCIPAL_TYPE_GOOGLE_GROUP, DataAccessEntitlement.entitlement == c.ENTITLEMENT_READ, EmailGroup.google_group_id.is_not(None), _failed_apply_effect_matches_current_binding_intent(), ] if retryable_only: conditions.append( DataAccessCompiledEffect.status.in_( (EFFECT_STATUS_FAILED, EFFECT_STATUS_RETRYABLE), ), ) if search: like = f"%{search.strip()}%" conditions.append( or_( DataAccessResource.project_id.ilike(like), DataAccessResource.dataset_id.ilike(like), DataAccessResource.display_name.ilike(like), EmailGroup.email.ilike(like), DataAccessCompiledEffect.provider_member.ilike(like), DataAccessCompiledEffect.last_error.ilike(like), ), ) return conditions def _failed_apply_effect_matches_current_binding_intent(): return or_( and_( DataAccessBinding.status == BINDING_STATUS_ACTIVE, DataAccessCompiledEffect.desired_state == EFFECT_DESIRED_PRESENT, ), and_( DataAccessBinding.status.in_( (BINDING_STATUS_REVOKED_PENDING, BINDING_STATUS_REVOKED), ), DataAccessCompiledEffect.desired_state == EFFECT_DESIRED_ABSENT, ), ) def _reconcile_run_conditions( *, search: str | None, mode: str | None, effect_status: str | None, ): conditions = [ DataAccessEffectRun.provider == c.PROVIDER_BIGQUERY, DataAccessEffectRun.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEffectRun.provider_member.like( f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}%", ), DataAccessEffectRun.provider_role == c.BIGQUERY_DATA_VIEWER_ROLE, DataAccessEffectRun.desired_state.in_( (EFFECT_DESIRED_PRESENT, EFFECT_DESIRED_ABSENT), ), DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEntitlement.domain == c.DOMAIN_DATA_ACCESS, DataAccessEntitlement.provider == c.PROVIDER_BIGQUERY, DataAccessEntitlement.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEntitlement.principal_type == c.PRINCIPAL_TYPE_GOOGLE_GROUP, DataAccessEntitlement.entitlement == c.ENTITLEMENT_READ, EmailGroup.google_group_id.is_not(None), ] clean_mode = mode.strip() if mode else None if clean_mode: conditions.append(DataAccessEffectRun.mode == clean_mode) clean_status = effect_status.strip() if effect_status else None if clean_status: conditions.append(DataAccessEffectRun.effect_status == clean_status) if search: like = f"%{search.strip()}%" conditions.append( or_( DataAccessResource.project_id.ilike(like), DataAccessResource.dataset_id.ilike(like), DataAccessResource.display_name.ilike(like), EmailGroup.email.ilike(like), DataAccessEffectRun.provider_member.ilike(like), DataAccessEffectRun.provider_outcome.ilike(like), DataAccessEffectRun.error_message.ilike(like), DataAccessEffectRun.trace_id.ilike(like), cast(DataAccessEffectRun.run_id, String).ilike(like), ), ) return conditions def _reconcile_run_row_select(): return ( select( DataAccessEffectRun, DataAccessCompiledEffect, DataAccessBinding, DataAccessResource, EmailGroup, ) .join( DataAccessCompiledEffect, DataAccessCompiledEffect.id == DataAccessEffectRun.effect_id, ) .join( DataAccessBinding, DataAccessBinding.id == DataAccessEffectRun.binding_id, ) .join( DataAccessResource, DataAccessResource.id == DataAccessEffectRun.resource_id, ) .join( DataAccessEntitlement, DataAccessEntitlement.id == DataAccessCompiledEffect.entitlement_id, ) .join( EmailGroup, EmailGroup.id == DataAccessCompiledEffect.principal_email_group_id, ) ) async def _reconcile_run_summary( db: AsyncSession, *, conditions, total: int, ) -> DataAccessReconcileRunSummaryResponse: rows = ( await db.execute( select( DataAccessEffectRun.effect_status, DataAccessEffectRun.desired_state, func.count(), ) .select_from(DataAccessEffectRun) .join( DataAccessCompiledEffect, DataAccessCompiledEffect.id == DataAccessEffectRun.effect_id, ) .join( DataAccessBinding, DataAccessBinding.id == DataAccessEffectRun.binding_id, ) .join( DataAccessResource, DataAccessResource.id == DataAccessEffectRun.resource_id, ) .join( DataAccessEntitlement, DataAccessEntitlement.id == DataAccessCompiledEffect.entitlement_id, ) .join( EmailGroup, EmailGroup.id == DataAccessCompiledEffect.principal_email_group_id, ) .where(*conditions) .group_by( DataAccessEffectRun.effect_status, DataAccessEffectRun.desired_state, ), ) ).all() return DataAccessReconcileRunSummaryResponse( total=total, failed=sum( int(count) for status, _, count in rows if status == EFFECT_STATUS_FAILED ), retryable=sum( int(count) for status, _, count in rows if status == EFFECT_STATUS_RETRYABLE ), applied=sum( int(count) for status, _, count in rows if status == EFFECT_STATUS_APPLIED ), desired_present=sum( int(count) for _, desired_state, count in rows if desired_state == EFFECT_DESIRED_PRESENT ), desired_absent=sum( int(count) for _, desired_state, count in rows if desired_state == EFFECT_DESIRED_ABSENT ), ) def _reconcile_run_item_response( *, run: DataAccessEffectRun, effect: DataAccessCompiledEffect, binding: DataAccessBinding, resource: DataAccessResource, principal: EmailGroup, ) -> DataAccessReconcileRunItemResponse: return DataAccessReconcileRunItemResponse( id=run.id, run_id=run.run_id, effect_id=effect.id, binding_id=binding.id, request_id=run.request_id, resource_id=resource.id, integration_id=run.integration_id, mode=run.mode, provider=run.provider, resource_type=run.resource_type, project_id=run.target_project_id, dataset_id=run.target_dataset_id, locator=f"{run.target_project_id}.{run.target_dataset_id}", display_name=resource.display_name, principal_email_group_id=principal.id, principal_email=principal.email, principal_google_group_id=principal.google_group_id, provider_member=run.provider_member, provider_role=run.provider_role, desired_state=run.desired_state, effect_status=run.effect_status, provider_outcome=run.provider_outcome, provider_etag=run.provider_etag, error_message=run.error_message, trace_id=run.trace_id, started_at=run.started_at, completed_at=run.completed_at, current_effect_status=effect.status, binding_status=binding.status, binding_reconcile_status=binding.reconcile_status, current_provider_outcome=effect.last_provider_outcome, current_error=effect.last_error or binding.last_reconcile_error, created_at=run.created_at, ) def _failed_apply_row_select(): return ( select( DataAccessCompiledEffect, DataAccessBinding, DataAccessResource, DataAccessEntitlement, EmailGroup, ) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .join( DataAccessResource, DataAccessResource.id == DataAccessCompiledEffect.resource_id, ) .join( DataAccessEntitlement, DataAccessEntitlement.id == DataAccessCompiledEffect.entitlement_id, ) .join( EmailGroup, EmailGroup.id == DataAccessCompiledEffect.principal_email_group_id, ) ) async def _failed_apply_summary( db: AsyncSession, *, conditions, total: int, ) -> DataAccessFailedApplySummaryResponse: rows = ( await db.execute( select( DataAccessCompiledEffect.status, DataAccessCompiledEffect.desired_state, func.count(), ) .select_from(DataAccessCompiledEffect) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .join( DataAccessResource, DataAccessResource.id == DataAccessCompiledEffect.resource_id, ) .join( DataAccessEntitlement, DataAccessEntitlement.id == DataAccessCompiledEffect.entitlement_id, ) .join( EmailGroup, EmailGroup.id == DataAccessCompiledEffect.principal_email_group_id, ) .where(*conditions) .group_by( DataAccessCompiledEffect.status, DataAccessCompiledEffect.desired_state, ), ) ).all() return DataAccessFailedApplySummaryResponse( total=total, failed=sum( int(count) for status, _, count in rows if status == EFFECT_STATUS_FAILED ), retryable=sum( int(count) for status, _, count in rows if status == EFFECT_STATUS_RETRYABLE ), desired_present=sum( int(count) for _, desired_state, count in rows if desired_state == EFFECT_DESIRED_PRESENT ), desired_absent=sum( int(count) for _, desired_state, count in rows if desired_state == EFFECT_DESIRED_ABSENT ), ) async def _latest_failed_apply_runs( db: AsyncSession, effect_ids: list[UUID], ) -> dict[UUID, DataAccessEffectRun]: if not effect_ids: return {} runs = ( await db.scalars( select(DataAccessEffectRun) .where(DataAccessEffectRun.effect_id.in_(effect_ids)) .order_by(*_failed_apply_run_order()), ) ).all() latest: dict[UUID, DataAccessEffectRun] = {} for run in runs: latest.setdefault(run.effect_id, run) return latest async def _latest_failed_apply_run( db: AsyncSession, effect_id: UUID, ) -> DataAccessEffectRun | None: return await db.scalar( select(DataAccessEffectRun) .where(DataAccessEffectRun.effect_id == effect_id) .order_by(*_failed_apply_run_order()) .limit(1), ) def _failed_apply_run_order(): admin_retry_rank = case( (DataAccessEffectRun.mode == "admin_retry", 1), else_=0, ) return ( DataAccessEffectRun.completed_at.desc(), admin_retry_rank.asc(), DataAccessEffectRun.created_at.desc(), DataAccessEffectRun.id.desc(), ) async def _load_failed_apply_row_for_retry( db: AsyncSession, effect_id: UUID, ): conditions = _failed_apply_conditions(search=None, retryable_only=True) return ( await db.execute( _failed_apply_row_select() .where(DataAccessCompiledEffect.id == effect_id, *conditions) .with_for_update(), ) ).one_or_none() def _record_failed_apply_retry_requested( db: AsyncSession, *, effect: DataAccessCompiledEffect, binding: DataAccessBinding, resource: DataAccessResource, retry_run_id: UUID, trace_id: str, started_at: datetime, ) -> None: db.add( DataAccessEffectRun( run_id=retry_run_id, effect_id=effect.id, binding_id=binding.id, request_id=binding.source_request_id, resource_id=resource.id, integration_id=str(resource.integration_id), trace_id=trace_id, mode="admin_retry", provider=effect.provider, resource_type=effect.resource_type, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, effect_status=effect.status, provider_outcome="retry_requested", provider_etag=effect.last_provider_etag, error_message=effect.last_error, started_at=started_at, completed_at=datetime.now(UTC), ), ) def _failed_apply_item_response( *, effect: DataAccessCompiledEffect, binding: DataAccessBinding, resource: DataAccessResource, entitlement: DataAccessEntitlement, principal: EmailGroup, latest_run: DataAccessEffectRun | None, ) -> DataAccessFailedApplyItemResponse: return DataAccessFailedApplyItemResponse( effect_id=effect.id, binding_id=binding.id, resource_id=resource.id, request_id=binding.source_request_id, integration_id=str(resource.integration_id), project_id=resource.project_id, dataset_id=resource.dataset_id, locator=f"{resource.project_id}.{resource.dataset_id}", display_name=resource.display_name, principal_email_group_id=principal.id, principal_email=principal.email, principal_google_group_id=str(principal.google_group_id), entitlement=entitlement.entitlement, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, managed_reason=effect.managed_reason, status=effect.status, binding_status=binding.status, reconcile_status=binding.reconcile_status, last_provider_outcome=effect.last_provider_outcome, last_provider_etag=effect.last_provider_etag, last_error=effect.last_error, latest_run=( DataAccessFailedApplyLatestRunResponse( id=latest_run.id, run_id=latest_run.run_id, mode=latest_run.mode, effect_status=latest_run.effect_status, provider_outcome=latest_run.provider_outcome, provider_etag=latest_run.provider_etag, error_message=latest_run.error_message, trace_id=latest_run.trace_id, completed_at=latest_run.completed_at, ) if latest_run else None ), created_at=effect.created_at, updated_at=effect.updated_at, last_reconcile_at=binding.last_reconcile_at, ) def _cutover_validation_item_response( item: DataAccessCutoverReportItem, ) -> DataAccessCutoverValidationItemResponse: return DataAccessCutoverValidationItemResponse( resource_id=item.resource_id, integration_id=item.integration_id, provider=item.provider, resource_type=item.resource_type, project_id=item.project_id, dataset_id=item.dataset_id, locator=item.locator, display_name=item.display_name, catalog_resource_id=item.catalog_resource_id, status=item.status, requestable=item.requestable, owner_source=item.owner_source, owner_person_id=item.owner_person_id, owner_resolution_status=item.owner_resolution_status, source_version=item.source_version, source_etag=item.source_etag, findings=list(item.findings), blocking_findings=list(item.blocking_findings), invalid_reasons=list(item.invalid_reasons), history=DataAccessCutoverValidationHistoryResponse( request_item_count=item.history.request_item_count, binding_count=item.history.binding_count, active_binding_count=item.history.active_binding_count, revoked_binding_count=item.history.revoked_binding_count, absent_effect_count=item.history.absent_effect_count, effect_run_count=item.history.effect_run_count, has_preserved_history=item.history.has_preserved_history, ), ) async def _require_data_access_request_author( db: AsyncSession, current_user: AuthUser, ) -> None: # Datasets are the shared selection surface for Group, Service Account, # and authenticated-person requests. Group principals still require the # owner/member authorization in ``authorize_data_access_requester`` at # submit time, but a person must not need to own a Group merely to browse # datasets they may request for themselves. person = await db.get(IdentityPerson, current_user.id) if person is not None and person.employment_status == "active": return raise AppException( code="DATA_ACCESS_ACTIVE_EMPLOYEE_REQUIRED", message=( "Only active employees may browse requestable data-access datasets." ), status_code=403, ) async def _data_access_member_group_ids_for_user( db: AsyncSession, current_user: AuthUser, ) -> set[UUID]: person = await db.get(IdentityPerson, current_user.id) if person is None: return set() override = person.__dict__.get("_data_access_member_group_ids_override") if override is not None: return { value if isinstance(value, UUID) else UUID(str(value)) for value in override } try: result = await EmailGroupService(db).list_member_groups_for_person( person=person, ) except Exception: # noqa: BLE001 # Membership is an *additive* enrichment on top of group-ownership; # it must never make the browse/authorization path hard-depend on # live Google Directory uptime. If the lookup fails (Google down / # rate-limited / not configured), degrade to owner-only rather than # 500 every non-admin — including pure owners who never needed it. logger.warning( "data-access member-group lookup failed for person=%s; " "degrading to owner-only", current_user.id, exc_info=True, ) return set() return { group.group_id for group in result.groups if group.group_id is not None } async def _resolve_data_access_group_access_reader( db: AsyncSession, current_user: AuthUser, principal_email_group_id: UUID, ) -> tuple[EmailGroup, bool, int]: principal = await db.get(EmailGroup, principal_email_group_id) if principal is None or principal.google_group_id is None: _raise_group_access_not_found() is_global_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE) if principal.status != "active": has_data_access_binding = await db.scalar( select(DataAccessBinding.id) .where(DataAccessBinding.principal_email_group_id == principal.id) .limit(1), ) if not is_global_admin or has_data_access_binding is None: _raise_group_access_not_found() owner_count = int( await db.scalar( select(func.count()) .select_from(EmailGroupOwner) .where(EmailGroupOwner.email_group_id == principal.id), ) or 0 ) caller_is_owner = bool( await db.scalar( select(EmailGroupOwner.id) .where( EmailGroupOwner.email_group_id == principal.id, EmailGroupOwner.person_id == current_user.id, ) .limit(1), ) ) if is_global_admin or (principal.status == "active" and caller_is_owner): return principal, caller_is_owner, owner_count _raise_group_access_not_found() def _raise_group_access_not_found() -> NoReturn: """Always raises — see ``_raise_request_not_found`` for why the ``NoReturn`` annotation matters (it is what lets a checker narrow the ``principal`` binding after the guard, instead of leaving five ``EmailGroup | None`` attribute accesses unverified).""" raise AppException( code="DATA_ACCESS_GROUP_ACCESS_NOT_FOUND", message="Data-access Google Group access not found.", status_code=404, ) class _DataAccessRequestDetailScope: def __init__(self, visible_resource_ids: set[UUID] | None = None) -> None: self.visible_resource_ids = visible_resource_ids @property def full_request(self) -> bool: return self.visible_resource_ids is None async def _resolve_data_access_request_detail_reader_scope( db: AsyncSession, current_user: AuthUser, request_row: DataAccessRequest, ) -> _DataAccessRequestDetailScope: if has_role(current_user.role_names, GLOBAL_ADMIN_ROLE): return _DataAccessRequestDetailScope() if str(request_row.requester_person_id) == str(current_user.id): return _DataAccessRequestDetailScope() if ( request_row.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT and request_row.principal_id is not None ): owns_service_account = await db.scalar( select(DataAccessServiceAccountProfile.principal_id).where( DataAccessServiceAccountProfile.principal_id == request_row.principal_id, DataAccessServiceAccountProfile.technical_owner_person_id == current_user.id, ).limit(1), ) if owns_service_account is not None: return _DataAccessRequestDetailScope() owns_group = await db.scalar( select(EmailGroupOwner.id) .where( EmailGroupOwner.email_group_id == request_row.principal_email_group_id, EmailGroupOwner.person_id == current_user.id, ) .limit(1), ) if owns_group is not None: return _DataAccessRequestDetailScope() owned_resource_ids = set( ( await db.scalars( select(DataAccessResource.id) .join( DataAccessRequestItem, DataAccessRequestItem.resource_id == DataAccessResource.id, ) .where( DataAccessRequestItem.request_id == request_row.id, DataAccessResource.owner_person_id == str(current_user.id), DataAccessResource.owner_resolution_status == OWNER_RESOLUTION_RESOLVED, ) ) ).all(), ) if owned_resource_ids: return _DataAccessRequestDetailScope( visible_resource_ids=owned_resource_ids, ) _raise_request_not_found() def _scope_request_bindings_query( *, request_id: UUID, visible_item_ids: set[UUID], detail_scope: _DataAccessRequestDetailScope, ): query = select(DataAccessBinding) if not detail_scope.full_request: query = query.where( DataAccessBinding.source_request_item_id.in_(visible_item_ids), ) return query.where(DataAccessBinding.source_request_id == request_id) def _scope_request_effects_query( *, visible_item_ids: set[UUID], detail_scope: _DataAccessRequestDetailScope, ): query = select(DataAccessCompiledEffect) if not detail_scope.full_request: query = query.where( DataAccessBinding.source_request_item_id.in_(visible_item_ids), ) return query def _scope_request_effect_runs_query( *, request_id: UUID, visible_effect_ids: set[UUID], detail_scope: _DataAccessRequestDetailScope, ): query = select(DataAccessEffectRun).where( DataAccessEffectRun.request_id == request_id, ) if not detail_scope.full_request: query = query.where(DataAccessEffectRun.effect_id.in_(visible_effect_ids)) return query def _raise_request_not_found() -> NoReturn: """Always raises. Annotated ``NoReturn`` rather than ``None`` on purpose: every caller is ``if row is None: _raise_request_not_found()``, and with ``-> None`` a type checker has to assume control can fall through, so ``row`` stays ``Model | None`` for the whole rest of the function. That single wrong annotation was the sole cause of **34** ``union-attr`` errors in this file — i.e. 34 places where nothing was verifying the None-safety of a ``request_row.`` access. """ raise AppException( code="DATA_ACCESS_REQUEST_NOT_FOUND", message="Data-access request not found.", status_code=404, ) def _group_access_item_response( *, binding: DataAccessBinding, resource: DataAccessResource, entitlement: DataAccessEntitlement, owner_label: str | None, effect: DataAccessCompiledEffect | None, latest_run: DataAccessEffectRun | None, ) -> DataAccessGroupAccessItemResponse: access_state = _group_access_state(binding=binding, effect=effect) latest_error = ( effect.last_error if effect and effect.last_error else binding.last_reconcile_error or (latest_run.error_message if latest_run else None) ) return DataAccessGroupAccessItemResponse( binding_id=binding.id, resource_id=resource.id, source_request_id=binding.source_request_id, resource_status=resource.status, resource_requestable=resource.requestable, project_id=resource.project_id, dataset_id=resource.dataset_id, locator=f"{resource.project_id}.{resource.dataset_id}", display_name=resource.display_name, owner_person_id=( str(resource.owner_person_id) if resource.owner_person_id else None ), owner_label=owner_label, owner_resolution_status=resource.owner_resolution_status, entitlement=binding.entitlement_snapshot, entitlement_version=binding.entitlement_version_snapshot, principal_type=entitlement.principal_type, provider_member=( effect.provider_member if effect else f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{binding.principal_email_snapshot}" ), provider_role=( effect.provider_role if effect else entitlement.provider_role ), binding_status=binding.status, revoke_reason=binding.revoke_reason, access_state=access_state, notice_code=_group_access_notice_code(binding=binding, resource=resource), reconcile_status=binding.reconcile_status, desired_state=effect.desired_state if effect else None, effect_status=effect.status if effect else None, latest_provider_outcome=( effect.last_provider_outcome if effect and effect.last_provider_outcome else latest_run.provider_outcome if latest_run else None ), latest_error=latest_error, latest_run=( DataAccessGroupAccessLatestRunResponse( id=latest_run.id, run_id=latest_run.run_id, mode=latest_run.mode, effect_status=latest_run.effect_status, provider_outcome=latest_run.provider_outcome, provider_etag=latest_run.provider_etag, error_message=latest_run.error_message, trace_id=latest_run.trace_id, completed_at=latest_run.completed_at, ) if latest_run else None ), already_present_at_first_apply=( effect.already_present_at_first_apply if effect else None ), granted_at=binding.granted_at, revoked_at=binding.revoked_at, last_reconcile_at=binding.last_reconcile_at, ) def _requestable_resource_availability(resource: DataAccessResource) -> str: if resource.status == RESOURCE_STATUS_SUSPENDED: return DATA_ACCESS_RESOURCE_AVAILABILITY_CATALOG_MISSING_SUSPENDED if resource.status == RESOURCE_STATUS_ACTIVE and resource.requestable: return DATA_ACCESS_RESOURCE_AVAILABILITY_REQUESTABLE return DATA_ACCESS_RESOURCE_AVAILABILITY_NOT_REQUESTABLE def _requestable_resource_notice_code(resource: DataAccessResource) -> str | None: if resource.status == RESOURCE_STATUS_SUSPENDED: return DATA_ACCESS_NOTICE_CATALOG_MISSING_SUSPENSION return None def _group_access_notice_code( *, binding: DataAccessBinding, resource: DataAccessResource, ) -> str | None: if binding.revoke_reason == DATA_ACCESS_REVOKE_REASON_GROUP_DELETED: return DATA_ACCESS_NOTICE_GROUP_DELETED_REVOCATION if resource.status == RESOURCE_STATUS_SUSPENDED: return DATA_ACCESS_NOTICE_CATALOG_MISSING_SUSPENSION return None def _group_access_state( *, binding: DataAccessBinding, effect: DataAccessCompiledEffect | None, ) -> str: return binding_access_state(binding, effect) def _group_access_item_sort_key( item: DataAccessGroupAccessItemResponse, ) -> tuple[int, str, str, str]: state_rank = { "active": 0, "pending_apply": 1, "failed": 2, "revoked_pending": 3, "revoked_history": 4, } return ( state_rank.get(item.access_state, 9), item.project_id, item.dataset_id, str(item.binding_id), ) def _group_access_summary( items: list[DataAccessGroupAccessItemResponse], ) -> DataAccessGroupAccessSummaryResponse: return DataAccessGroupAccessSummaryResponse( total=len(items), active=sum(1 for item in items if item.access_state == "active"), pending_apply=sum( 1 for item in items if item.access_state == "pending_apply" ), failed=sum(1 for item in items if item.access_state == "failed"), revoked_pending=sum( 1 for item in items if item.binding_status == BINDING_STATUS_REVOKED_PENDING ), revoked_history=sum( 1 for item in items if item.binding_status == BINDING_STATUS_REVOKED ), ) def _detail_item_response( *, item: DataAccessRequestItem, resource: DataAccessResource, request_row: DataAccessRequest, entitlement: DataAccessEntitlement | None, owner_label: str | None, binding: DataAccessBinding | None, effect: DataAccessCompiledEffect | None, runs_by_effect: dict[UUID, list[DataAccessEffectRun]], ) -> DataAccessRequestDetailItemResponse: preview = ( _preview_from_effect(effect) if effect is not None else _preview_from_request_item( item=item, request_row=request_row, entitlement=entitlement, resource=resource, ) ) return DataAccessRequestDetailItemResponse( id=item.id, line_no=item.line_no, resource_id=item.resource_id, project_id=item.resource_project_id_snapshot, dataset_id=item.resource_dataset_id_snapshot, table_id=resource.table_id, locator=( f"{item.resource_project_id_snapshot}." f"{item.resource_dataset_id_snapshot}" f"{'.' + resource.table_id if resource.table_id else ''}" ), display_name=resource.display_name, owner_person_id=str(resource.owner_person_id) if resource.owner_person_id else None, owner_label=owner_label, owner_resolution_status=resource.owner_resolution_status, item_status=item.status, display_state=_detail_item_display_state( request_status=request_row.status, item_status=item.status, binding=binding, effect=effect, ), staleness_reason=item.staleness_reason, source_version_snapshot=item.resource_source_version_snapshot, source_etag_snapshot=item.resource_source_etag_snapshot, binding_id=binding.id if binding else None, binding_status=binding.status if binding else None, binding_reconcile_status=binding.reconcile_status if binding else None, binding_last_reconcile_at=binding.last_reconcile_at if binding else None, binding_last_reconcile_error=binding.last_reconcile_error if binding else None, compiled_preview=preview, effect=( _effect_response(effect, runs_by_effect.get(effect.id, [])) if effect else None ), ) def _preview_from_request_item( *, item: DataAccessRequestItem, request_row: DataAccessRequest, entitlement: DataAccessEntitlement | None, resource: DataAccessResource, ) -> DataAccessRequestDetailCompiledPreviewResponse: return DataAccessRequestDetailCompiledPreviewResponse( provider=c.PROVIDER_BIGQUERY, resource_type=resource.resource_type, target_project_id=item.resource_project_id_snapshot, target_dataset_id=item.resource_dataset_id_snapshot, target_table_id=resource.table_id, provider_member=( f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}" f"{request_row.principal_email_snapshot}" ), provider_role=( entitlement.provider_role if entitlement else c.BIGQUERY_DATA_VIEWER_ROLE ), desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=request_row.entitlement_version_snapshot, resource_source_version_snapshot=item.resource_source_version_snapshot, resource_source_etag_snapshot=item.resource_source_etag_snapshot, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, ) def _preview_from_effect( effect: DataAccessCompiledEffect, ) -> DataAccessRequestDetailCompiledPreviewResponse: return DataAccessRequestDetailCompiledPreviewResponse( provider=effect.provider, resource_type=effect.resource_type, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, target_table_id=effect.target_table_id, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, entitlement_version_snapshot=effect.entitlement_version_snapshot, resource_source_version_snapshot=effect.resource_source_version_snapshot, resource_source_etag_snapshot=effect.resource_source_etag_snapshot, compiler_version=effect.compiler_version, managed_reason=effect.managed_reason, already_present_at_first_apply=effect.already_present_at_first_apply, ) def _effect_response( effect: DataAccessCompiledEffect, runs: list[DataAccessEffectRun], ) -> DataAccessRequestDetailEffectResponse: return DataAccessRequestDetailEffectResponse( **_preview_from_effect(effect).model_dump(), id=effect.id, binding_id=effect.binding_id, status=effect.status, last_provider_etag=effect.last_provider_etag, last_provider_outcome=effect.last_provider_outcome, last_error=effect.last_error, updated_at=effect.updated_at, runs=[ DataAccessRequestDetailRunResponse( id=run.id, run_id=run.run_id, mode=run.mode, effect_status=run.effect_status, provider_outcome=run.provider_outcome, provider_etag=run.provider_etag, error_message=run.error_message, trace_id=run.trace_id, started_at=run.started_at, completed_at=run.completed_at, ) for run in runs ], ) def _detail_item_display_state( *, request_status: str, item_status: str, binding: DataAccessBinding | None, effect: DataAccessCompiledEffect | None, ) -> str: if request_status == REQUEST_STATUS_REJECTED or item_status == REQUEST_ITEM_STATUS_REJECTED: return "rejected" if ( request_status == REQUEST_STATUS_STALE_REQUIRES_RESUBMIT or item_status == REQUEST_ITEM_STATUS_STALE ): return "stale" if effect is not None and effect.status in { EFFECT_STATUS_FAILED, EFFECT_STATUS_RETRYABLE, }: return "failed" if binding is not None and binding.reconcile_status in { RECONCILE_STATUS_FAILED, RECONCILE_STATUS_RETRYABLE, }: return "failed" if effect is not None and effect.status == EFFECT_STATUS_APPLIED: return "active" if binding is not None and binding.reconcile_status == RECONCILE_STATUS_APPLIED: return "active" if ( request_status == REQUEST_STATUS_APPROVED or item_status == REQUEST_ITEM_STATUS_APPROVED or binding is not None or ( effect is not None and effect.status == EFFECT_STATUS_PENDING_RECONCILE ) ): return "approved_pending_apply" return "pending" def _visible_items_reconcile_status( items: list[DataAccessRequestDetailItemResponse], ) -> str: if not items: return RECONCILE_STATUS_NOT_STARTED states = {item.display_state for item in items} if "failed" in states: return RECONCILE_STATUS_RETRYABLE if states == {"active"}: return RECONCILE_STATUS_APPLIED if "approved_pending_apply" in states: return RECONCILE_STATUS_PENDING return RECONCILE_STATUS_NOT_STARTED def _visible_items_last_run_id( items: list[DataAccessRequestDetailItemResponse], ) -> UUID | None: runs = [ run for item in items if item.effect is not None for run in item.effect.runs ] if not runs: return None return max(runs, key=lambda run: run.completed_at).run_id def _visible_items_last_reconcile_at( items: list[DataAccessRequestDetailItemResponse], ) -> datetime | None: timestamps = [ value for item in items for value in ( item.binding_last_reconcile_at, item.effect.updated_at if item.effect else None, ) if value is not None ] return max(timestamps) if timestamps else None def _visible_items_last_error( items: list[DataAccessRequestDetailItemResponse], ) -> str | None: for item in items: if item.effect and item.effect.last_error: return item.effect.last_error if item.binding_last_reconcile_error: return item.binding_last_reconcile_error return None