"""Catalog feed projection for data-access BigQuery datasets. The catalog feed is ingestion input only. This module validates catalog-shaped dataset entries and projects them into ``data_access_resources``; request submission, Feishu callbacks, and revoke paths keep reading that row instead of raw catalog payloads. """ from __future__ import annotations import hashlib import json import re import uuid from collections.abc import Iterable, Mapping from dataclasses import dataclass from typing import Any, Final from sqlalchemy import case, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.domains.data_access import constants as c from app.domains.data_access.models import ( OWNER_RESOLUTION_FALLBACK, OWNER_RESOLUTION_RESOLVED, OWNER_RESOLUTION_STATUSES, OWNER_RESOLUTION_UNRESOLVED, OWNER_SOURCE_BOOTSTRAP, OWNER_SOURCE_BOOTSTRAP_TRANSFERRED, OWNER_SOURCE_CATALOG, OWNER_SOURCE_CATALOG_TRANSFERRED, OWNER_SOURCE_DATASET_INHERITED, RESOURCE_STATUS_ACTIVE, RESOURCE_STATUS_DELETED, RESOURCE_STATUS_SUSPENDED, RESOURCE_STATUSES, DataAccessBinding, DataAccessCompiledEffect, DataAccessEffectRun, DataAccessRequestItem, DataAccessResource, ) from app.domains.data_access.service import ( DATA_ACCESS_RESOURCE_REMOVAL_REASON_PROVIDER_DELETED, DataAccessService, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration CATALOG_OWNER_SOURCE: Final = OWNER_SOURCE_CATALOG OWNER_RESOLUTION_STATUS_FIELD: Final = "owner_resolution_status" FIELD_LOCATOR: Final = "locator" FIELD_CATALOG_ID: Final = "catalog_id" FIELD_DISPLAY_NAME: Final = "display_name" FIELD_DESCRIPTION: Final = "description" FIELD_LABELS: Final = "labels" FIELD_REQUESTABLE: Final = "requestable" FIELD_STATUS: Final = "status" CATALOG_FEED_OUTCOME_CREATED: Final = "created" CATALOG_FEED_OUTCOME_UPDATED: Final = "updated" CATALOG_FEED_OUTCOME_UNCHANGED: Final = "unchanged" CATALOG_FEED_OUTCOME_BOOTSTRAP_REPLACED: Final = "bootstrap_replaced" CATALOG_FEED_OUTCOME_BOOTSTRAP_OWNER_MISMATCH_SUSPENDED: Final = ( "bootstrap_owner_mismatch_suspended" ) CATALOG_FEED_OUTCOME_BOOTSTRAP_LOCATOR_DRIFT_SUSPENDED: Final = ( "bootstrap_locator_drift_suspended" ) CATALOG_FEED_OUTCOME_CATALOG_LOCATOR_DRIFT_SUSPENDED: Final = ( "catalog_locator_drift_suspended" ) CATALOG_FEED_OUTCOME_CATALOG_MISSING_SUSPENDED: Final = ( "catalog_missing_suspended" ) _PROJECT_ID_RE = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$") _DATASET_ID_RE = re.compile(r"^[A-Za-z0-9_]{1,1024}$") _ENTRY_KEYS: Final = frozenset({ "provider", "resource_type", FIELD_LOCATOR, FIELD_CATALOG_ID, c.FIELD_OWNER_PERSON_ID, OWNER_RESOLUTION_STATUS_FIELD, c.FIELD_SOURCE_VERSION, c.FIELD_SOURCE_ETAG, FIELD_DISPLAY_NAME, FIELD_DESCRIPTION, FIELD_LABELS, FIELD_REQUESTABLE, FIELD_STATUS, }) _LOCATOR_KEYS: Final = frozenset({"project_id", "dataset_id"}) _FORBIDDEN_SHAPE_KEYS: Final = frozenset({ "table", "table_id", "table_ref", "dataset", "dataset_ref", "project", "project_id", "project_number", "resource", "resource_name", "fully_qualified_name", }) _FORBIDDEN_LOCATOR_KEYS: Final = _FORBIDDEN_SHAPE_KEYS - _LOCATOR_KEYS _MAX_LABEL_KEY_LENGTH: Final = 128 _MAX_LABEL_VALUE_LENGTH: Final = 512 @dataclass(frozen=True, slots=True) class CatalogDatasetLocator: """Canonical provider locator for one catalog-fed BigQuery dataset.""" project_id: str dataset_id: str @property def key(self) -> tuple[str, str]: return (self.project_id, self.dataset_id) @dataclass(frozen=True, slots=True) class CatalogDatasetResource: """Normalized catalog projection for one data-access resource row.""" provider: str resource_type: str locator: CatalogDatasetLocator owner_person_id: str | None owner_resolution_status: str source_version: str source_etag: str catalog_id: str | None = None display_name: str | None = None description: str | None = None labels: dict[str, str] | None = None requestable: bool = True status: str = RESOURCE_STATUS_ACTIVE @property def resource_owner_authority_enabled(self) -> bool: return ( self.owner_resolution_status == OWNER_RESOLUTION_RESOLVED and self.owner_person_id is not None ) def as_resource_projection(self) -> dict[str, Any]: return { "provider": self.provider, "resource_type": self.resource_type, "project_id": self.locator.project_id, "dataset_id": self.locator.dataset_id, "catalog_resource_id": self.catalog_id, "display_name": self.display_name, "notes": self.description, "labels": self.labels, "requestable": self.requestable, "status": self.status, c.FIELD_OWNER_PERSON_ID: self.owner_person_id, OWNER_RESOLUTION_STATUS_FIELD: self.owner_resolution_status, c.FIELD_OWNER_SOURCE: OWNER_SOURCE_CATALOG, c.FIELD_SOURCE_VERSION: self.source_version, c.FIELD_SOURCE_ETAG: self.source_etag, "resource_owner_authority_enabled": ( self.resource_owner_authority_enabled ), } @dataclass(frozen=True, slots=True) class CatalogFeedProjectionOutcome: """Per-resource projection result returned to feed callers and audit.""" resource_id: uuid.UUID outcome: str project_id: str dataset_id: str owner_person_id: str | None owner_resolution_status: str requestable: bool status: str source_version: str source_etag: str catalog_id: str | None = None changed: bool = False metadata: dict[str, Any] | None = None def as_audit_detail(self) -> dict[str, Any]: detail = { "resource_id": str(self.resource_id), "outcome": self.outcome, "project_id": self.project_id, "dataset_id": self.dataset_id, "catalog_id": self.catalog_id, "owner_person_id": self.owner_person_id, "owner_resolution_status": self.owner_resolution_status, "requestable": self.requestable, "status": self.status, "source_version": self.source_version, "source_etag": self.source_etag, "changed": self.changed, } if self.metadata: detail["metadata"] = self.metadata return detail @dataclass(frozen=True, slots=True) class CatalogFeedProjectionResult: """Batch projection summary for a catalog feed upsert.""" integration_id: str entry_count: int created_count: int updated_count: int unchanged_count: int bootstrap_replaced_count: int owner_mismatch_count: int locator_drift_count: int suspended_count: int outcomes: tuple[CatalogFeedProjectionOutcome, ...] class DataAccessCatalogFeedError(AppException): """Raised when a catalog feed cannot be projected safely.""" def __init__( self, reason: str, message: str, *, field: str = "catalog_feed", status_code: int = 422, ) -> None: self.reason = reason super().__init__( code=f"DATA_ACCESS_CATALOG_FEED_{reason.upper()}", message=message, status_code=status_code, field=field, ) def parse_catalog_resource(raw: Mapping[str, Any]) -> CatalogDatasetResource: """Validate and normalize one catalog-fed BigQuery dataset entry.""" if not isinstance(raw, Mapping): raise ValueError("catalog resource must be an object") _validate_entry_keys(raw) locator = _parse_dataset_locator(raw.get(FIELD_LOCATOR)) provider = c.assert_canonical("provider", _required_string(raw, "provider")) resource_type = c.assert_canonical( "resource_type", _required_string(raw, "resource_type"), ) if resource_type != c.RESOURCE_TYPE_DATASET: raise ValueError("catalog feed accepts BigQuery dataset resources only") owner_person_id = _parse_owner_person_id(raw.get(c.FIELD_OWNER_PERSON_ID)) owner_resolution_status = _parse_owner_resolution_status( raw.get(OWNER_RESOLUTION_STATUS_FIELD), owner_person_id=owner_person_id, ) requestable = raw.get(FIELD_REQUESTABLE, True) if not isinstance(requestable, bool): raise ValueError("requestable must be a boolean when provided") status = raw.get(FIELD_STATUS, RESOURCE_STATUS_ACTIVE) if not isinstance(status, str) or status not in RESOURCE_STATUSES: raise ValueError(f"status must be one of {sorted(RESOURCE_STATUSES)!r}") if status == RESOURCE_STATUS_DELETED: requestable = False return CatalogDatasetResource( provider=provider, resource_type=resource_type, locator=locator, owner_person_id=owner_person_id, owner_resolution_status=owner_resolution_status, source_version=_required_string(raw, c.FIELD_SOURCE_VERSION), source_etag=_required_string(raw, c.FIELD_SOURCE_ETAG), catalog_id=_optional_string(raw.get(FIELD_CATALOG_ID)), display_name=_optional_string(raw.get(FIELD_DISPLAY_NAME)), description=_optional_string(raw.get(FIELD_DESCRIPTION)), labels=_parse_labels(raw.get(FIELD_LABELS)), requestable=requestable, status=status, ) def parse_catalog_resources( entries: Iterable[Mapping[str, Any]], ) -> tuple[CatalogDatasetResource, ...]: """Validate a catalog feed batch and reject duplicate active locators.""" parsed: list[CatalogDatasetResource] = [] seen: set[tuple[str, str, str]] = set() seen_catalog_ids: set[tuple[str, str, str]] = set() for index, entry in enumerate(entries): try: resource = parse_catalog_resource(entry) except ValueError as exc: raise ValueError(f"catalog resource #{index}: {exc}") from exc key = ( resource.provider, resource.locator.project_id, resource.locator.dataset_id, ) if key in seen: raise ValueError( "duplicate catalog BigQuery dataset locator: " f"{resource.locator.project_id}.{resource.locator.dataset_id}", ) seen.add(key) if resource.catalog_id is not None: catalog_key = ( resource.provider, resource.resource_type, resource.catalog_id, ) if catalog_key in seen_catalog_ids: raise ValueError( "duplicate catalog BigQuery catalog id: " f"{resource.catalog_id}", ) seen_catalog_ids.add(catalog_key) parsed.append(resource) return tuple(parsed) async def project_catalog_dataset_feed( db: AsyncSession, *, integration_id: str | uuid.UUID, entries: Iterable[Mapping[str, Any] | CatalogDatasetResource], trace_id: str | uuid.UUID | None = None, actor_user_id: uuid.UUID | None = None, ) -> CatalogFeedProjectionResult: """Upsert catalog-fed BigQuery datasets into ``data_access_resources``. Existing-row identity prefers stable catalog id when provided, then falls back to ``integration_id + provider + resource_type + project_id + dataset_id``. The function never calls BigQuery and never creates requests, bindings, effects, or Feishu approvals. """ integration = await _load_data_access_integration(db, integration_id) resources = _coerce_catalog_resources(entries) await _validate_resolved_owners_exist(db, resources) outcomes: list[CatalogFeedProjectionOutcome] = [] for resource in resources: row, match_strategy = await _find_existing_resource( db, integration_id=str(integration.id), resource=resource, ) digest = _source_payload_digest(resource) if row is None: row = DataAccessResource( integration_id=str(integration.id), provider=resource.provider, resource_type=resource.resource_type, project_id=resource.locator.project_id, dataset_id=resource.locator.dataset_id, display_name=resource.display_name, notes=resource.description, labels=resource.labels, catalog_resource_id=resource.catalog_id, requestable=resource.requestable, status=resource.status, owner_person_id=resource.owner_person_id, owner_resolution_status=resource.owner_resolution_status, owner_source=OWNER_SOURCE_CATALOG, source_version=resource.source_version, source_etag=resource.source_etag, source_payload_digest=digest, ) db.add(row) await db.flush() outcome = CATALOG_FEED_OUTCOME_CREATED outcome_changed = True outcome_metadata = None elif row.owner_source == OWNER_SOURCE_BOOTSTRAP: has_preserved_history = ( await _has_preserved_resource_history(db, row.id) if _locator_changed(row, resource) else False ) outcome, outcome_changed, outcome_metadata = ( _apply_bootstrap_cutover_projection( row, resource, digest=digest, match_strategy=match_strategy or "provider_locator", has_preserved_history=has_preserved_history, ) ) await db.flush() elif row.owner_source == OWNER_SOURCE_BOOTSTRAP_TRANSFERRED: row.owner_source = OWNER_SOURCE_CATALOG_TRANSFERRED changed = _apply_resource_projection(row, resource, digest=digest) await db.flush() outcome = ( CATALOG_FEED_OUTCOME_UPDATED if changed else CATALOG_FEED_OUTCOME_UNCHANGED ) outcome_changed = changed outcome_metadata = None else: has_preserved_history = ( await _has_preserved_resource_history(db, row.id) if _should_suspend_catalog_id_locator_drift( row, resource, match_strategy=match_strategy, ) else False ) if has_preserved_history: outcome_metadata = _projection_metadata( row, resource, match_strategy=match_strategy or "provider_locator", has_preserved_history=True, ) outcome_changed = _suspend_resource_locator_drift_for_admin_resolution( row, resource, digest=digest, ) await db.flush() outcome = CATALOG_FEED_OUTCOME_CATALOG_LOCATOR_DRIFT_SUSPENDED else: changed = _apply_resource_projection(row, resource, digest=digest) await db.flush() outcome = ( CATALOG_FEED_OUTCOME_UPDATED if changed else CATALOG_FEED_OUTCOME_UNCHANGED ) outcome_changed = changed outcome_metadata = None await _propagate_dataset_projection_to_table_resources(db, row) outcomes.append( CatalogFeedProjectionOutcome( resource_id=row.id, outcome=outcome, project_id=row.project_id, dataset_id=row.dataset_id, catalog_id=row.catalog_resource_id, owner_person_id=row.owner_person_id, owner_resolution_status=row.owner_resolution_status, requestable=row.requestable, status=row.status, source_version=row.source_version, source_etag=row.source_etag, changed=outcome_changed, metadata=outcome_metadata, ), ) if resource.status == RESOURCE_STATUS_DELETED: await DataAccessService().confirm_data_access_resource_removed( db, resource_id=row.id, reason=DATA_ACCESS_RESOURCE_REMOVAL_REASON_PROVIDER_DELETED, trace_id=( str(trace_id) if trace_id is not None else f"data_access_catalog_removed:{row.id}" ), actor_type="integration", actor_user_id=actor_user_id, ) outcomes.extend( await _suspend_missing_catalog_resources( db, integration_id=str(integration.id), present_resources=resources, ), ) result = _projection_result( str(integration.id), entry_count=len(resources), outcomes=outcomes, ) await _audit_catalog_projection( db, result, trace_id=trace_id, actor_user_id=actor_user_id, ) await db.flush() return result async def _propagate_dataset_projection_to_table_resources( db: AsyncSession, dataset: DataAccessResource, ) -> None: """Keep materialized table rows aligned with their parent catalog row. Table rows own only table-specific delegation. Their provider locator and catalog provenance remain derived from the parent dataset. An explicitly delegated table owner is preserved; an inherited owner follows catalog owner changes. """ tables = ( await db.scalars( select(DataAccessResource).where( DataAccessResource.parent_resource_id == dataset.id, DataAccessResource.resource_type == c.RESOURCE_TYPE_TABLE, ), ) ).all() for table in tables: table.integration_id = dataset.integration_id table.project_id = dataset.project_id table.dataset_id = dataset.dataset_id table.requestable = dataset.requestable table.status = dataset.status table.source_version = dataset.source_version table.source_etag = dataset.source_etag table.source_payload_digest = dataset.source_payload_digest if table.owner_source == OWNER_SOURCE_DATASET_INHERITED: table.owner_person_id = dataset.owner_person_id table.owner_resolution_status = dataset.owner_resolution_status if tables: await db.flush() def _coerce_catalog_resources( entries: Iterable[Mapping[str, Any] | CatalogDatasetResource], ) -> tuple[CatalogDatasetResource, ...]: resources: list[CatalogDatasetResource] = [] for index, entry in enumerate(entries): try: resources.append(_coerce_catalog_resource(entry)) except ValueError as exc: raise DataAccessCatalogFeedError( "invalid_resource", f"Catalog feed entry #{index} is invalid: {exc}", ) from exc return _reject_duplicate_resources(resources) def _coerce_catalog_resource( entry: Mapping[str, Any] | CatalogDatasetResource, ) -> CatalogDatasetResource: if isinstance(entry, CatalogDatasetResource): return parse_catalog_resource(_preconstructed_resource_to_raw(entry)) return parse_catalog_resource(entry) def _preconstructed_resource_to_raw( resource: CatalogDatasetResource, ) -> dict[str, Any]: if not isinstance(resource.locator, CatalogDatasetLocator): raise ValueError("locator must be a CatalogDatasetLocator") return { "provider": resource.provider, "resource_type": resource.resource_type, FIELD_LOCATOR: { "project_id": resource.locator.project_id, "dataset_id": resource.locator.dataset_id, }, FIELD_CATALOG_ID: resource.catalog_id, c.FIELD_OWNER_PERSON_ID: resource.owner_person_id, OWNER_RESOLUTION_STATUS_FIELD: resource.owner_resolution_status, c.FIELD_SOURCE_VERSION: resource.source_version, c.FIELD_SOURCE_ETAG: resource.source_etag, FIELD_DISPLAY_NAME: resource.display_name, FIELD_DESCRIPTION: resource.description, FIELD_LABELS: resource.labels, FIELD_REQUESTABLE: resource.requestable, FIELD_STATUS: resource.status, } def _reject_duplicate_resources( resources: Iterable[CatalogDatasetResource], ) -> tuple[CatalogDatasetResource, ...]: parsed: list[CatalogDatasetResource] = [] seen: set[tuple[str, str, str]] = set() seen_catalog_ids: set[tuple[str, str, str]] = set() for resource in resources: key = ( resource.provider, resource.locator.project_id, resource.locator.dataset_id, ) if key in seen: raise DataAccessCatalogFeedError( "duplicate_locator", "Catalog feed contains duplicate BigQuery dataset locators.", field="locator", ) seen.add(key) if resource.catalog_id is not None: catalog_key = ( resource.provider, resource.resource_type, resource.catalog_id, ) if catalog_key in seen_catalog_ids: raise DataAccessCatalogFeedError( "duplicate_catalog_id", "Catalog feed contains duplicate BigQuery catalog ids.", field=FIELD_CATALOG_ID, ) seen_catalog_ids.add(catalog_key) parsed.append(resource) return tuple(parsed) async def _load_data_access_integration( db: AsyncSession, integration_id: str | uuid.UUID, ) -> Integration: try: integration_uuid = _parse_uuid(str(integration_id), field="integration_id") except ValueError as exc: raise DataAccessCatalogFeedError( "invalid_integration_id", "Catalog feed integration_id must be a UUID.", field="integration_id", ) from exc integration = await db.get(Integration, integration_uuid) if integration is None: raise DataAccessCatalogFeedError( "unknown_integration", "Data-access catalog feed integration is unknown.", field="integration_id", status_code=404, ) if integration.kind != c.DOMAIN_DATA_ACCESS or integration.provider != ( c.PROVIDER_BIGQUERY ): raise DataAccessCatalogFeedError( "unsupported_integration", "Catalog feed projection requires a data_access/bigquery integration.", field="integration_id", ) return integration async def _validate_resolved_owners_exist( db: AsyncSession, resources: tuple[CatalogDatasetResource, ...], ) -> None: owner_ids = { uuid.UUID(resource.owner_person_id) for resource in resources if resource.owner_person_id is not None } if not owner_ids: return existing = set( await db.scalars( select(IdentityPerson.id).where(IdentityPerson.id.in_(owner_ids)), ), ) missing = sorted(str(owner_id) for owner_id in owner_ids - existing) if missing: raise DataAccessCatalogFeedError( "unknown_owner", "Catalog feed references owner_person_id values Helix does not know.", field=c.FIELD_OWNER_PERSON_ID, ) async def _find_existing_resource( db: AsyncSession, *, integration_id: str, resource: CatalogDatasetResource, ) -> tuple[DataAccessResource | None, str | None]: if resource.catalog_id is not None: rows = ( await db.scalars( select(DataAccessResource) .where( DataAccessResource.integration_id == integration_id, DataAccessResource.provider == resource.provider, DataAccessResource.resource_type == resource.resource_type, DataAccessResource.catalog_resource_id == resource.catalog_id, ) .order_by( case( (DataAccessResource.status == RESOURCE_STATUS_ACTIVE, 0), else_=1, ), DataAccessResource.updated_at.desc(), DataAccessResource.id, ) .limit(2), ) ).all() if len(rows) > 1: raise DataAccessCatalogFeedError( "duplicate_catalog_id", "Data-access resources contain duplicate catalog ids.", field=FIELD_CATALOG_ID, ) if rows: return rows[0], "catalog_id" row = await db.scalar( select(DataAccessResource) .where( DataAccessResource.integration_id == integration_id, DataAccessResource.provider == resource.provider, DataAccessResource.resource_type == resource.resource_type, DataAccessResource.project_id == resource.locator.project_id, DataAccessResource.dataset_id == resource.locator.dataset_id, ) .order_by( case( (DataAccessResource.status == RESOURCE_STATUS_ACTIVE, 0), else_=1, ), DataAccessResource.updated_at.desc(), DataAccessResource.id, ) .limit(1), ) if row is not None: return row, "provider_locator" return None, None def _apply_resource_projection( row: DataAccessResource, resource: CatalogDatasetResource, *, digest: str, ) -> bool: updates: dict[str, Any] = { "project_id": resource.locator.project_id, "dataset_id": resource.locator.dataset_id, "display_name": resource.display_name, "notes": resource.description, "labels": resource.labels, "catalog_resource_id": resource.catalog_id, "requestable": resource.requestable, "status": resource.status, "source_version": resource.source_version, "source_etag": resource.source_etag, "source_payload_digest": digest, } if row.owner_source != OWNER_SOURCE_CATALOG_TRANSFERRED: updates.update({ "owner_person_id": resource.owner_person_id, "owner_resolution_status": resource.owner_resolution_status, "owner_source": OWNER_SOURCE_CATALOG, }) changed = False for field, value in updates.items(): if getattr(row, field) != value: setattr(row, field, value) changed = True return changed def _apply_bootstrap_cutover_projection( row: DataAccessResource, resource: CatalogDatasetResource, *, digest: str, match_strategy: str, has_preserved_history: bool, ) -> tuple[str, bool, dict[str, Any] | None]: metadata = _projection_metadata( row, resource, match_strategy=match_strategy, has_preserved_history=has_preserved_history, ) if _bootstrap_owner_matches_catalog(row, resource): if ( match_strategy == "catalog_id" and has_preserved_history and _locator_changed(row, resource) ): changed = _suspend_resource_locator_drift_for_admin_resolution( row, resource, digest=digest, ) return ( CATALOG_FEED_OUTCOME_BOOTSTRAP_LOCATOR_DRIFT_SUSPENDED, changed, metadata, ) changed = _apply_resource_projection(row, resource, digest=digest) return ( CATALOG_FEED_OUTCOME_BOOTSTRAP_REPLACED if changed else CATALOG_FEED_OUTCOME_UNCHANGED, changed, metadata, ) updates: dict[str, Any] = { "display_name": resource.display_name, "notes": resource.description, "labels": resource.labels, "catalog_resource_id": resource.catalog_id, "requestable": False, "status": RESOURCE_STATUS_SUSPENDED, "owner_person_id": None, "owner_resolution_status": OWNER_RESOLUTION_UNRESOLVED, "source_payload_digest": digest, } changed = False for field, value in updates.items(): if getattr(row, field) != value: setattr(row, field, value) changed = True return ( CATALOG_FEED_OUTCOME_BOOTSTRAP_OWNER_MISMATCH_SUSPENDED, changed, metadata, ) def _projection_metadata( row: DataAccessResource, resource: CatalogDatasetResource, *, match_strategy: str, has_preserved_history: bool, ) -> dict[str, Any]: return { "previous_owner_person_id": row.owner_person_id, "previous_owner_resolution_status": row.owner_resolution_status, "catalog_owner_person_id": resource.owner_person_id, "catalog_owner_resolution_status": resource.owner_resolution_status, "previous_locator": f"{row.project_id}.{row.dataset_id}", "catalog_locator": ( f"{resource.locator.project_id}.{resource.locator.dataset_id}" ), "catalog_source_version": resource.source_version, "catalog_source_etag": resource.source_etag, "match_strategy": match_strategy, "preserved_history_present": has_preserved_history, "provider_mutation_attempted": False, } def _suspend_resource_locator_drift_for_admin_resolution( row: DataAccessResource, resource: CatalogDatasetResource, *, digest: str, ) -> bool: updates: dict[str, Any] = { "display_name": resource.display_name, "notes": resource.description, "labels": resource.labels, "catalog_resource_id": resource.catalog_id, "requestable": False, "status": RESOURCE_STATUS_SUSPENDED, "source_payload_digest": digest, } changed = False for field, value in updates.items(): if getattr(row, field) != value: setattr(row, field, value) changed = True return changed def _locator_changed( row: DataAccessResource, resource: CatalogDatasetResource, ) -> bool: return ( row.project_id != resource.locator.project_id or row.dataset_id != resource.locator.dataset_id ) def _should_suspend_catalog_id_locator_drift( row: DataAccessResource, resource: CatalogDatasetResource, *, match_strategy: str | None, ) -> bool: return ( match_strategy == "catalog_id" and resource.status != RESOURCE_STATUS_DELETED and _locator_changed(row, resource) ) async def _has_preserved_resource_history( db: AsyncSession, resource_id: uuid.UUID, ) -> bool: history_queries = ( select(DataAccessRequestItem.id) .where(DataAccessRequestItem.resource_id == resource_id) .limit(1), select(DataAccessBinding.id) .where(DataAccessBinding.resource_id == resource_id) .limit(1), select(DataAccessCompiledEffect.id) .where(DataAccessCompiledEffect.resource_id == resource_id) .limit(1), select(DataAccessEffectRun.id) .where(DataAccessEffectRun.resource_id == resource_id) .limit(1), ) for query in history_queries: if await db.scalar(query) is not None: return True return False def _bootstrap_owner_matches_catalog( row: DataAccessResource, resource: CatalogDatasetResource, ) -> bool: return ( row.owner_resolution_status == OWNER_RESOLUTION_RESOLVED and resource.owner_resolution_status == OWNER_RESOLUTION_RESOLVED and row.owner_person_id == resource.owner_person_id ) async def _suspend_missing_catalog_resources( db: AsyncSession, *, integration_id: str, present_resources: tuple[CatalogDatasetResource, ...], ) -> list[CatalogFeedProjectionOutcome]: present_keys = { ( resource.provider, resource.resource_type, resource.locator.project_id, resource.locator.dataset_id, ) for resource in present_resources } active_catalog_rows = ( await db.scalars( select(DataAccessResource).where( DataAccessResource.integration_id == integration_id, DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessResource.owner_source.in_(( OWNER_SOURCE_CATALOG, OWNER_SOURCE_CATALOG_TRANSFERRED, )), DataAccessResource.status == RESOURCE_STATUS_ACTIVE, ), ) ).all() outcomes: list[CatalogFeedProjectionOutcome] = [] for row in active_catalog_rows: key = (row.provider, row.resource_type, row.project_id, row.dataset_id) if key in present_keys: continue row.status = RESOURCE_STATUS_SUSPENDED outcomes.append( CatalogFeedProjectionOutcome( resource_id=row.id, outcome=CATALOG_FEED_OUTCOME_CATALOG_MISSING_SUSPENDED, project_id=row.project_id, dataset_id=row.dataset_id, owner_person_id=row.owner_person_id, owner_resolution_status=row.owner_resolution_status, requestable=row.requestable, status=row.status, source_version=row.source_version, source_etag=row.source_etag, catalog_id=row.catalog_resource_id, changed=True, ), ) if outcomes: await db.flush() return outcomes def _projection_result( integration_id: str, *, entry_count: int, outcomes: list[CatalogFeedProjectionOutcome], ) -> CatalogFeedProjectionResult: return CatalogFeedProjectionResult( integration_id=integration_id, entry_count=entry_count, created_count=sum( outcome.outcome == CATALOG_FEED_OUTCOME_CREATED for outcome in outcomes ), updated_count=sum( outcome.changed and outcome.outcome != CATALOG_FEED_OUTCOME_CREATED for outcome in outcomes ), unchanged_count=sum( outcome.outcome == CATALOG_FEED_OUTCOME_UNCHANGED for outcome in outcomes ), bootstrap_replaced_count=sum( outcome.outcome == CATALOG_FEED_OUTCOME_BOOTSTRAP_REPLACED for outcome in outcomes ), owner_mismatch_count=sum( outcome.outcome == CATALOG_FEED_OUTCOME_BOOTSTRAP_OWNER_MISMATCH_SUSPENDED for outcome in outcomes ), locator_drift_count=sum( outcome.outcome in { CATALOG_FEED_OUTCOME_BOOTSTRAP_LOCATOR_DRIFT_SUSPENDED, CATALOG_FEED_OUTCOME_CATALOG_LOCATOR_DRIFT_SUSPENDED, } for outcome in outcomes ), suspended_count=sum( outcome.outcome == CATALOG_FEED_OUTCOME_CATALOG_MISSING_SUSPENDED for outcome in outcomes ), outcomes=tuple(outcomes), ) async def _audit_catalog_projection( db: AsyncSession, result: CatalogFeedProjectionResult, *, trace_id: str | uuid.UUID | None, actor_user_id: uuid.UUID | None, ) -> None: await AuditService(db).log( trace_id=trace_id or f"data_access_catalog_feed:{result.integration_id}", actor_type="integration", actor_user_id=actor_user_id, action="data_access.catalog_feed.project", status="success", resource_type="data_access_integration", resource_id=result.integration_id, detail={ "entry_count": result.entry_count, "created_count": result.created_count, "updated_count": result.updated_count, "unchanged_count": result.unchanged_count, "bootstrap_replaced_count": result.bootstrap_replaced_count, "owner_mismatch_count": result.owner_mismatch_count, "locator_drift_count": result.locator_drift_count, "suspended_count": result.suspended_count, "provider_mutation_attempted": False, "request_handlers_read_raw_catalog": False, "feishu_callbacks_read_raw_catalog": False, "outcomes": [outcome.as_audit_detail() for outcome in result.outcomes], }, ) def _source_payload_digest(resource: CatalogDatasetResource) -> str: payload = resource.as_resource_projection() payload.pop("resource_owner_authority_enabled", None) serialized = json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ) return f"sha256:{hashlib.sha256(serialized.encode()).hexdigest()}" def _validate_entry_keys(raw: Mapping[str, Any]) -> None: keys = frozenset(str(k) for k in raw.keys()) forbidden = keys & _FORBIDDEN_SHAPE_KEYS if forbidden: raise ValueError( "catalog data-access entries must be dataset resources; " "top-level table/project/freeform fields are forbidden: " f"{sorted(forbidden)!r}", ) extra = keys - _ENTRY_KEYS if extra: raise ValueError( "unexpected catalog data-access fields: " f"{sorted(extra)!r}", ) def _parse_dataset_locator(raw: Any) -> CatalogDatasetLocator: if not isinstance(raw, Mapping): raise ValueError( "locator must be an object with project_id and dataset_id; " "freeform dataset strings are not accepted", ) keys = frozenset(str(k) for k in raw.keys()) forbidden = keys & _FORBIDDEN_LOCATOR_KEYS if forbidden: raise ValueError( "catalog data-access locators are dataset-scoped only; " f"forbidden locator fields: {sorted(forbidden)!r}", ) extra = keys - _LOCATOR_KEYS if extra: raise ValueError( "locator must contain only project_id and dataset_id; " f"unexpected fields: {sorted(extra)!r}", ) project_id = _required_string(raw, "project_id") dataset_id = _required_string(raw, "dataset_id") if not _PROJECT_ID_RE.fullmatch(project_id): raise ValueError( "locator.project_id must be a canonical GCP project id, " "not a freeform project entry", ) if "." in dataset_id or ":" in dataset_id or "/" in dataset_id: raise ValueError( "locator.dataset_id must name a dataset only, not a table path", ) if not _DATASET_ID_RE.fullmatch(dataset_id): raise ValueError("locator.dataset_id is not a valid BigQuery dataset id") return CatalogDatasetLocator(project_id=project_id, dataset_id=dataset_id) def _parse_owner_person_id(raw: Any) -> str | None: if raw in (None, ""): return None if not isinstance(raw, str): raise ValueError(f"{c.FIELD_OWNER_PERSON_ID} must be a UUID string") return str(_parse_uuid(raw.strip(), field=c.FIELD_OWNER_PERSON_ID)) def _parse_owner_resolution_status( raw: Any, *, owner_person_id: str | None, ) -> str: if owner_person_id is not None: if raw in (None, "", OWNER_RESOLUTION_RESOLVED): return OWNER_RESOLUTION_RESOLVED raise ValueError( f"{OWNER_RESOLUTION_STATUS_FIELD} cannot be {raw!r} when " f"{c.FIELD_OWNER_PERSON_ID} is present", ) status = _required_string( {OWNER_RESOLUTION_STATUS_FIELD: raw}, OWNER_RESOLUTION_STATUS_FIELD, ) if status == OWNER_RESOLUTION_RESOLVED: raise ValueError( f"{c.FIELD_OWNER_PERSON_ID} is required when " f"{OWNER_RESOLUTION_STATUS_FIELD} is {OWNER_RESOLUTION_RESOLVED!r}", ) if status not in OWNER_RESOLUTION_STATUSES: raise ValueError( f"{OWNER_RESOLUTION_STATUS_FIELD} must be one of " f"{sorted(OWNER_RESOLUTION_STATUSES)!r}", ) if status not in {OWNER_RESOLUTION_UNRESOLVED, OWNER_RESOLUTION_FALLBACK}: raise ValueError( f"{OWNER_RESOLUTION_STATUS_FIELD} does not grant owner authority " f"without {c.FIELD_OWNER_PERSON_ID}", ) return status def _parse_labels(raw: Any) -> dict[str, str] | None: if raw in (None, ""): return None if not isinstance(raw, Mapping): raise ValueError("labels must be an object when provided") labels: dict[str, str] = {} for key, value in raw.items(): if not isinstance(key, str) or not key.strip(): raise ValueError("labels keys must be non-empty strings") if not isinstance(value, str): raise ValueError("labels values must be strings") clean_key = key.strip() clean_value = value.strip() if len(clean_key) > _MAX_LABEL_KEY_LENGTH: raise ValueError("labels keys must be at most 128 characters") if len(clean_value) > _MAX_LABEL_VALUE_LENGTH: raise ValueError("labels values must be at most 512 characters") labels[clean_key] = clean_value return dict(sorted(labels.items())) or None def _required_string(raw: Mapping[str, Any], key: str) -> str: value = raw.get(key) if not isinstance(value, str) or not value.strip(): raise ValueError(f"{key} is required") return value.strip() def _optional_string(raw: Any) -> str | None: if raw in (None, ""): return None if not isinstance(raw, str): raise ValueError("optional string fields must be strings") return raw.strip() or None def _parse_uuid(raw: str, *, field: str) -> uuid.UUID: try: return uuid.UUID(str(raw).strip()) except ValueError as exc: raise ValueError(f"{field} must be a UUID") from exc