"""Observe-mode data-access reconciler. DAX-024 is intentionally read-only: it compares persisted compiled effects against the provider snapshot for the exact dataset locators represented by those effects. Provider mutation stays behind explicit apply paths. """ from __future__ import annotations import re import uuid from collections.abc import Sequence from dataclasses import dataclass from datetime import UTC, datetime from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload 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.apply_gate import ( DataAccessApplyGateConfig, data_access_apply_gate_config, dataset_locator_key, ) from app.domains.data_access.models import ( BINDING_STATUS_ACTIVE, BINDING_STATUS_REVOKED, BINDING_STATUS_REVOKED_PENDING, EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_PRESENT, EFFECT_STATUS_APPLIED, EFFECT_STATUS_FAILED, EFFECT_STATUS_OBSERVED, EFFECT_STATUS_PENDING_RECONCILE, EFFECT_STATUS_RETRYABLE, RECONCILE_STATUS_APPLIED, RECONCILE_STATUS_FAILED, RECONCILE_STATUS_NOT_STARTED, RECONCILE_STATUS_OBSERVED, RECONCILE_STATUS_PENDING, RECONCILE_STATUS_RETRYABLE, RESOURCE_STATUS_ACTIVE, DataAccessBinding, DataAccessCompiledEffect, DataAccessEffectRun, DataAccessRequest, DataAccessResource, ) from app.integrations.driver_base import ( DataAccessDatasetIamBinding, DataAccessDatasetIamPolicy, DataAccessDatasetIamPolicyPatch, DataAccessDatasetIamUpdateResult, DataAccessDriver, ) from app.integrations.models import Integration from app.integrations.service import build_driver_by_id OBSERVE_OUTCOME_UNCHANGED = "observe_unchanged" OBSERVE_OUTCOME_WOULD_ADD = "observe_would_add" OBSERVE_OUTCOME_WOULD_REMOVE = "observe_would_remove" OBSERVE_OUTCOME_ERROR = "observe_error" APPLY_GATE_REASON_ALLOWED = "allowed" APPLY_GATE_REASON_DISABLED = "apply_mode_disabled" APPLY_GATE_REASON_INVALID_CONFIG = "invalid_apply_gate_config" APPLY_GATE_REASON_RESOURCE_UNAVAILABLE = "resource_unavailable" APPLY_GATE_REASON_RESOURCE_LOCATOR_MISMATCH = "resource_locator_mismatch" APPLY_GATE_REASON_RESOURCE_INACTIVE = "resource_inactive" APPLY_GATE_REASON_RESOURCE_NOT_REQUESTABLE = "resource_not_requestable" APPLY_GATE_REASON_RESOURCE_OWNER_AUTHORITY_UNAVAILABLE = ( "resource_owner_authority_unavailable" ) APPLY_GATE_REASON_PERSONAL_DISABLED = "personal_apply_disabled" APPLY_GATE_REASON_SERVICE_ACCOUNT_DISABLED = "service_account_apply_disabled" APPLY_PRESENT_OUTCOME_APPLIED = "apply_present_applied" APPLY_PRESENT_OUTCOME_ALREADY_APPLIED = "apply_present_already_applied" APPLY_PRESENT_OUTCOME_CONFLICT = "apply_present_conflict" APPLY_PRESENT_OUTCOME_ERROR = "apply_present_error" APPLY_ABSENT_OUTCOME_APPLIED = "apply_absent_applied" APPLY_ABSENT_OUTCOME_ALREADY_ABSENT = "apply_absent_already_absent" APPLY_ABSENT_OUTCOME_CONFLICT = "apply_absent_conflict" APPLY_ABSENT_OUTCOME_ERROR = "apply_absent_error" _PRIVATE_KEY_RE = re.compile( r"-----BEGIN PRIVATE KEY-----.*?-----END PRIVATE KEY-----", re.DOTALL, ) @dataclass(frozen=True, slots=True) class DataAccessObserveEffectDelta: """One observed compiled effect delta.""" effect_id: uuid.UUID resource_id: uuid.UUID target_project_id: str target_dataset_id: str provider_member: str provider_role: str desired_state: str outcome: str provider_etag: str | None = None error_message: str | None = None target_table_id: str | None = None @dataclass(frozen=True, slots=True) class DataAccessUnmanagedProviderGrant: """One provider-side grant that Helix has no compiled effect for.""" resource_id: uuid.UUID target_project_id: str target_dataset_id: str provider_member: str provider_role: str provider_etag: str | None = None target_table_id: str | None = None @dataclass(frozen=True, slots=True) class DataAccessObserveReconcileResult: """Summary of one observe-mode reconciler run.""" run_id: uuid.UUID integration_id: str observed_effect_count: int would_add_count: int would_remove_count: int unchanged_count: int error_count: int unmanaged_grant_count: int deltas: tuple[DataAccessObserveEffectDelta, ...] unmanaged_grants: tuple[DataAccessUnmanagedProviderGrant, ...] @dataclass(frozen=True, slots=True) class DataAccessApplyGateDecision: """One pending effect's apply eligibility decision.""" effect_id: uuid.UUID resource_id: uuid.UUID target_project_id: str target_dataset_id: str provider_member: str principal_id: uuid.UUID | None principal_type_snapshot: str principal_subject_snapshot: str | None provider_access_kind: str provider_role: str desired_state: str locator: str allowed: bool reason: str target_table_id: str | None = None @dataclass(frozen=True, slots=True) class DataAccessApplyGateResult: """Apply-mode gate result for pending compiled effects.""" run_id: uuid.UUID integration_id: str apply_mode_enabled: bool selected_effect_count: int allowed_effect_count: int blocked_effect_count: int decisions: tuple[DataAccessApplyGateDecision, ...] config_error: str | None = None @dataclass(frozen=True, slots=True) class DataAccessApplyPresentEffectResult: """One present-effect provider mutation result.""" effect_id: uuid.UUID resource_id: uuid.UUID target_project_id: str target_dataset_id: str provider_member: str provider_role: str desired_state: str outcome: str already_present_at_first_apply: bool provider_etag: str | None = None error_message: str | None = None target_table_id: str | None = None @dataclass(frozen=True, slots=True) class DataAccessApplyPresentReconcileResult: """Summary of one apply-present reconciler run.""" run_id: uuid.UUID integration_id: str apply_mode_enabled: bool selected_effect_count: int selected_present_effect_count: int ignored_non_present_effect_count: int allowed_effect_count: int blocked_effect_count: int provider_write_count: int applied_count: int already_applied_count: int conflict_count: int error_count: int gate_decisions: tuple[DataAccessApplyGateDecision, ...] results: tuple[DataAccessApplyPresentEffectResult, ...] config_error: str | None = None @dataclass(frozen=True, slots=True) class DataAccessApplyAbsentEffectResult: """One absent-effect provider mutation result.""" effect_id: uuid.UUID resource_id: uuid.UUID target_project_id: str target_dataset_id: str provider_member: str provider_role: str desired_state: str outcome: str provider_etag: str | None = None error_message: str | None = None target_table_id: str | None = None @dataclass(frozen=True, slots=True) class DataAccessApplyAbsentReconcileResult: """Summary of one apply-absent reconciler run.""" run_id: uuid.UUID integration_id: str apply_mode_enabled: bool selected_effect_count: int selected_absent_effect_count: int ignored_non_absent_effect_count: int ignored_non_revoked_binding_effect_count: int allowed_effect_count: int blocked_effect_count: int provider_write_count: int removed_count: int already_absent_count: int conflict_count: int error_count: int gate_decisions: tuple[DataAccessApplyGateDecision, ...] results: tuple[DataAccessApplyAbsentEffectResult, ...] config_error: str | None = None async def observe_reconcile_compiled_effects( db: AsyncSession, *, integration_id: str | uuid.UUID, effect_ids: Sequence[uuid.UUID] | None = None, resource_ids: Sequence[uuid.UUID] | None = None, driver: DataAccessDriver | None = None, trace_id: str | uuid.UUID | None = None, ) -> DataAccessObserveReconcileResult: """Compare pending compiled effects with provider state, without applying. Only pending or retryable effects for the explicit data-access integration and optional caller-selected effect/resource ids are selected. The row selection uses ``FOR UPDATE SKIP LOCKED`` so concurrent workers partition the same bounded set instead of racing it. The runner groups those locked effects by their compiled dataset locator and calls ``get_dataset_iam`` once per locator. It never calls ``set_dataset_iam``. """ integration_id_text = str(integration_id) run_id = uuid.uuid4() started_at = datetime.now(UTC) trace_id_text = _trace_id_text(trace_id, integration_id_text) scoped_effect_ids = None if effect_ids is None else tuple(effect_ids) scoped_resource_ids = None if resource_ids is None else tuple(resource_ids) effects = await _load_pending_effects( db, integration_id=integration_id_text, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) if not effects: result = DataAccessObserveReconcileResult( run_id=run_id, integration_id=integration_id_text, observed_effect_count=0, would_add_count=0, would_remove_count=0, unchanged_count=0, error_count=0, unmanaged_grant_count=0, deltas=(), unmanaged_grants=(), ) await _audit_observe_result( db, result, trace_id=trace_id, run_id=run_id, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) return result data_access_driver = driver if data_access_driver is None: resolved = await build_driver_by_id(integration_id_text, session=db) if not isinstance(resolved, DataAccessDriver): raise AppException( code="DATA_ACCESS_DRIVER_INVALID", message=( "Configured data-access integration does not resolve to a " "DataAccessDriver." ), status_code=503, ) data_access_driver = resolved managed_members_by_locator = await _load_managed_members_by_locator(db, effects) deltas: list[DataAccessObserveEffectDelta] = [] unmanaged_grants: list[DataAccessUnmanagedProviderGrant] = [] for locator, locator_effects in _group_effects_by_locator(effects).items(): project_id, dataset_id, table_id = locator try: policy = await _get_locator_iam( data_access_driver, project_id, dataset_id, table_id, ) except Exception as exc: # noqa: BLE001 - driver/SDK errors vary error_message = _truncate_error(exc) for effect in locator_effects: effect.status = EFFECT_STATUS_RETRYABLE effect.last_provider_etag = None effect.last_provider_outcome = OBSERVE_OUTCOME_ERROR effect.last_error = error_message deltas.append(_delta_for_effect( effect, outcome=OBSERVE_OUTCOME_ERROR, provider_etag=None, error_message=error_message, )) _record_effect_run( db, effect=effect, integration_id=integration_id_text, run_id=run_id, trace_id=trace_id_text, outcome=OBSERVE_OUTCOME_ERROR, provider_etag=None, error_message=error_message, started_at=started_at, ) continue provider_members = _members_by_role(policy) unmanaged_grants.extend(_unmanaged_provider_grants( locator_effects, policy=policy, managed_provider_grants=managed_members_by_locator.get(locator, set()), )) for effect in locator_effects: outcome = _observe_outcome(effect, provider_members) effect.status = EFFECT_STATUS_OBSERVED effect.last_provider_etag = policy.etag effect.last_provider_outcome = outcome effect.last_error = None deltas.append(_delta_for_effect( effect, outcome=outcome, provider_etag=policy.etag, )) _record_effect_run( db, effect=effect, integration_id=integration_id_text, run_id=run_id, trace_id=trace_id_text, outcome=outcome, provider_etag=policy.etag, error_message=None, started_at=started_at, ) result = DataAccessObserveReconcileResult( run_id=run_id, integration_id=integration_id_text, observed_effect_count=len(deltas), would_add_count=sum( 1 for delta in deltas if delta.outcome == OBSERVE_OUTCOME_WOULD_ADD ), would_remove_count=sum( 1 for delta in deltas if delta.outcome == OBSERVE_OUTCOME_WOULD_REMOVE ), unchanged_count=sum( 1 for delta in deltas if delta.outcome == OBSERVE_OUTCOME_UNCHANGED ), error_count=sum( 1 for delta in deltas if delta.outcome == OBSERVE_OUTCOME_ERROR ), unmanaged_grant_count=len(unmanaged_grants), deltas=tuple(deltas), unmanaged_grants=tuple(unmanaged_grants), ) await _audit_observe_result( db, result, trace_id=trace_id, run_id=run_id, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) await db.flush() await _refresh_reconcile_rollups( db, effects=effects, run_id=run_id, observed_at=datetime.now(UTC), ) await db.flush() return result async def resolve_apply_reconcile_gate( db: AsyncSession, *, integration_id: str | uuid.UUID, effect_ids: Sequence[uuid.UUID] | None = None, resource_ids: Sequence[uuid.UUID] | None = None, trace_id: str | uuid.UUID | None = None, ) -> DataAccessApplyGateResult: """Resolve the apply gate without mutating provider or desired state. The dataset mutation boundary is enforced by Terraform-managed BigQuery IAM on the dedicated data-access service account. This gate only resolves the runtime apply-mode switch before provider writes are attempted. """ integration_id_text = str(integration_id) run_id = uuid.uuid4() scoped_effect_ids = None if effect_ids is None else tuple(effect_ids) scoped_resource_ids = None if resource_ids is None else tuple(resource_ids) gate_config, config_error = await _load_apply_gate_config( db, integration_id=integration_id_text, ) effects = await _load_pending_effects( db, integration_id=integration_id_text, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) decisions = tuple( _apply_gate_decision_for_effect( effect, gate_config=gate_config, config_error=config_error, ) for effect in effects ) result = DataAccessApplyGateResult( run_id=run_id, integration_id=integration_id_text, apply_mode_enabled=gate_config.apply_mode_enabled, selected_effect_count=len(decisions), allowed_effect_count=sum(1 for decision in decisions if decision.allowed), blocked_effect_count=sum(1 for decision in decisions if not decision.allowed), decisions=decisions, config_error=config_error, ) await _audit_apply_gate_result( db, result, trace_id=trace_id, run_id=run_id, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) return result async def apply_present_reconcile_compiled_effects( db: AsyncSession, *, integration_id: str | uuid.UUID, effect_ids: Sequence[uuid.UUID] | None = None, resource_ids: Sequence[uuid.UUID] | None = None, driver: DataAccessDriver | None = None, trace_id: str | uuid.UUID | None = None, ) -> DataAccessApplyPresentReconcileResult: """Apply allowed present effects through the BigQuery writer. DAX-033 is intentionally add-only. Absent/revoke effects stay pending for the DAX-034 slice. """ integration_id_text = str(integration_id) run_id = uuid.uuid4() started_at = datetime.now(UTC) trace_id_text = _trace_id_text( trace_id, integration_id_text, default_prefix="data_access_apply_present", ) scoped_effect_ids = None if effect_ids is None else tuple(effect_ids) scoped_resource_ids = None if resource_ids is None else tuple(resource_ids) gate_config, config_error = await _load_apply_gate_config( db, integration_id=integration_id_text, ) effects = await _load_pending_effects( db, integration_id=integration_id_text, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) present_effects = [ effect for effect in effects if effect.desired_state == EFFECT_DESIRED_PRESENT # Symmetric with the absent side's ``_is_revoked_binding_absent_effect`` # guard: never (re-)write a PRESENT effect whose binding has since been # revoked. A binding revoked while its PRESENT effect is still pending # only flips ``binding.status`` + queues an ABSENT effect (see # ``service._mark_binding_revoked_pending``) — it does NOT supersede the # in-flight PRESENT effect. Without this gate the next apply-present pass # would push the just-revoked group straight back into the dataset IAM # policy, re-granting data access Helix believes is gone. and _is_active_binding_present_effect(effect) ] gate_decisions = tuple( _apply_gate_decision_for_effect( effect, gate_config=gate_config, config_error=config_error, ) for effect in present_effects ) allowed_effect_ids = { decision.effect_id for decision in gate_decisions if decision.allowed } allowed_effects = [ effect for effect in present_effects if effect.id in allowed_effect_ids ] data_access_driver = driver if allowed_effects and data_access_driver is None: resolved = await build_driver_by_id(integration_id_text, session=db) if not isinstance(resolved, DataAccessDriver): raise AppException( code="DATA_ACCESS_DRIVER_INVALID", message=( "Configured data-access integration does not resolve to a " "DataAccessDriver." ), status_code=503, ) data_access_driver = resolved if allowed_effects and data_access_driver is None: raise AppException( code="DATA_ACCESS_DRIVER_INVALID", message="Data-access apply requires a DataAccessDriver.", status_code=503, ) results: list[DataAccessApplyPresentEffectResult] = [] provider_write_count = 0 for locator, locator_effects in _group_effects_by_locator(allowed_effects).items(): assert data_access_driver is not None project_id, dataset_id, table_id = locator try: policy = await _get_locator_iam( data_access_driver, project_id, dataset_id, table_id, ) except Exception as exc: # noqa: BLE001 - driver/SDK errors vary error_message = _truncate_error(exc) for effect in locator_effects: results.append(_record_apply_present_result( db, effect=effect, integration_id=integration_id_text, run_id=run_id, trace_id=trace_id_text, outcome=APPLY_PRESENT_OUTCOME_ERROR, status=EFFECT_STATUS_RETRYABLE, provider_etag=None, error_message=error_message, started_at=started_at, )) continue provider_members = _members_by_role(policy) patch = _present_effects_apply_patch(locator_effects) provider_write_count += 1 try: update = await _set_locator_iam( data_access_driver, project_id, dataset_id, table_id, policy.etag, patch, ) except Exception as exc: # noqa: BLE001 - driver/SDK errors vary update = DataAccessDatasetIamUpdateResult( ok=False, changed=False, error_message=_truncate_error(exc), ) for effect in locator_effects: results.append(_apply_present_result_for_update( db, effect=effect, integration_id=integration_id_text, run_id=run_id, trace_id=trace_id_text, update=update, fallback_etag=policy.etag, already_present_before_apply=_provider_has_effect( provider_members, effect, ), started_at=started_at, )) result = DataAccessApplyPresentReconcileResult( run_id=run_id, integration_id=integration_id_text, apply_mode_enabled=gate_config.apply_mode_enabled, selected_effect_count=len(effects), selected_present_effect_count=len(present_effects), ignored_non_present_effect_count=len(effects) - len(present_effects), allowed_effect_count=sum(1 for decision in gate_decisions if decision.allowed), blocked_effect_count=sum( 1 for decision in gate_decisions if not decision.allowed ), provider_write_count=provider_write_count, applied_count=sum( 1 for row in results if row.outcome == APPLY_PRESENT_OUTCOME_APPLIED ), already_applied_count=sum( 1 for row in results if row.outcome == APPLY_PRESENT_OUTCOME_ALREADY_APPLIED ), conflict_count=sum( 1 for row in results if row.outcome == APPLY_PRESENT_OUTCOME_CONFLICT ), error_count=sum( 1 for row in results if row.outcome == APPLY_PRESENT_OUTCOME_ERROR ), gate_decisions=gate_decisions, results=tuple(results), config_error=config_error, ) await _audit_apply_present_result( db, result, trace_id=trace_id, run_id=run_id, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) await db.flush() if results: await _refresh_reconcile_rollups( db, effects=allowed_effects, run_id=run_id, observed_at=datetime.now(UTC), ) await db.flush() return result async def apply_absent_reconcile_compiled_effects( db: AsyncSession, *, integration_id: str | uuid.UUID, effect_ids: Sequence[uuid.UUID] | None = None, resource_ids: Sequence[uuid.UUID] | None = None, driver: DataAccessDriver | None = None, trace_id: str | uuid.UUID | None = None, ) -> DataAccessApplyAbsentReconcileResult: """Apply allowed absent effects through the BigQuery writer. DAX-034 is remove-only. Present/grant effects stay pending for the apply-present runner. """ integration_id_text = str(integration_id) run_id = uuid.uuid4() started_at = datetime.now(UTC) trace_id_text = _trace_id_text( trace_id, integration_id_text, default_prefix="data_access_apply_absent", ) scoped_effect_ids = None if effect_ids is None else tuple(effect_ids) scoped_resource_ids = None if resource_ids is None else tuple(resource_ids) gate_config, config_error = await _load_apply_gate_config( db, integration_id=integration_id_text, ) effects = await _load_pending_effects( db, integration_id=integration_id_text, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) candidate_absent_effects = [ effect for effect in effects if effect.desired_state == EFFECT_DESIRED_ABSENT ] absent_effects = [ effect for effect in candidate_absent_effects if _is_revoked_binding_absent_effect(effect) ] gate_decisions = tuple( _apply_gate_decision_for_effect( effect, gate_config=gate_config, config_error=config_error, ) for effect in absent_effects ) allowed_effect_ids = { decision.effect_id for decision in gate_decisions if decision.allowed } allowed_effects = [ effect for effect in absent_effects if effect.id in allowed_effect_ids ] data_access_driver = driver if allowed_effects and data_access_driver is None: resolved = await build_driver_by_id(integration_id_text, session=db) if not isinstance(resolved, DataAccessDriver): raise AppException( code="DATA_ACCESS_DRIVER_INVALID", message=( "Configured data-access integration does not resolve to a " "DataAccessDriver." ), status_code=503, ) data_access_driver = resolved if allowed_effects and data_access_driver is None: raise AppException( code="DATA_ACCESS_DRIVER_INVALID", message="Data-access apply requires a DataAccessDriver.", status_code=503, ) results: list[DataAccessApplyAbsentEffectResult] = [] provider_write_count = 0 for locator, locator_effects in _group_effects_by_locator(allowed_effects).items(): assert data_access_driver is not None project_id, dataset_id, table_id = locator try: policy = await _get_locator_iam( data_access_driver, project_id, dataset_id, table_id, ) except Exception as exc: # noqa: BLE001 - driver/SDK errors vary error_message = _truncate_error(exc) for effect in locator_effects: results.append(_record_apply_absent_result( db, effect=effect, integration_id=integration_id_text, run_id=run_id, trace_id=trace_id_text, outcome=APPLY_ABSENT_OUTCOME_ERROR, status=EFFECT_STATUS_RETRYABLE, provider_etag=None, error_message=error_message, started_at=started_at, )) continue patch = _absent_effects_apply_patch(locator_effects) provider_write_count += 1 try: update = await _set_locator_iam( data_access_driver, project_id, dataset_id, table_id, policy.etag, patch, ) except Exception as exc: # noqa: BLE001 - driver/SDK errors vary update = DataAccessDatasetIamUpdateResult( ok=False, changed=False, error_message=_truncate_error(exc), ) for effect in locator_effects: results.append(_apply_absent_result_for_update( db, effect=effect, integration_id=integration_id_text, run_id=run_id, trace_id=trace_id_text, update=update, fallback_etag=policy.etag, started_at=started_at, )) result = DataAccessApplyAbsentReconcileResult( run_id=run_id, integration_id=integration_id_text, apply_mode_enabled=gate_config.apply_mode_enabled, selected_effect_count=len(effects), selected_absent_effect_count=len(absent_effects), ignored_non_absent_effect_count=len(effects) - len(candidate_absent_effects), ignored_non_revoked_binding_effect_count=( len(candidate_absent_effects) - len(absent_effects) ), allowed_effect_count=sum(1 for decision in gate_decisions if decision.allowed), blocked_effect_count=sum( 1 for decision in gate_decisions if not decision.allowed ), provider_write_count=provider_write_count, removed_count=sum( 1 for row in results if row.outcome == APPLY_ABSENT_OUTCOME_APPLIED ), already_absent_count=sum( 1 for row in results if row.outcome == APPLY_ABSENT_OUTCOME_ALREADY_ABSENT ), conflict_count=sum( 1 for row in results if row.outcome == APPLY_ABSENT_OUTCOME_CONFLICT ), error_count=sum( 1 for row in results if row.outcome == APPLY_ABSENT_OUTCOME_ERROR ), gate_decisions=gate_decisions, results=tuple(results), config_error=config_error, ) await _audit_apply_absent_result( db, result, trace_id=trace_id, run_id=run_id, effect_ids=scoped_effect_ids, resource_ids=scoped_resource_ids, ) await db.flush() if results: await _refresh_reconcile_rollups( db, effects=allowed_effects, run_id=run_id, observed_at=datetime.now(UTC), ) await db.flush() return result async def _load_pending_effects( db: AsyncSession, *, integration_id: str, effect_ids: tuple[uuid.UUID, ...] | None, resource_ids: tuple[uuid.UUID, ...] | None, ) -> list[DataAccessCompiledEffect]: if effect_ids == () or resource_ids == (): return [] resource_scope = select(DataAccessResource.id).where( DataAccessResource.integration_id == integration_id, DataAccessResource.provider == c.PROVIDER_BIGQUERY, DataAccessResource.resource_type.in_( (c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE), ), ) if resource_ids is not None: resource_scope = resource_scope.where(DataAccessResource.id.in_(resource_ids)) stmt = ( select(DataAccessCompiledEffect) .options( selectinload(DataAccessCompiledEffect.binding), selectinload(DataAccessCompiledEffect.resource), ) .where( DataAccessCompiledEffect.resource_id.in_(resource_scope), DataAccessCompiledEffect.status.in_( (EFFECT_STATUS_PENDING_RECONCILE, EFFECT_STATUS_RETRYABLE), ), DataAccessCompiledEffect.provider == c.PROVIDER_BIGQUERY, DataAccessCompiledEffect.resource_type.in_( (c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE), ), ) .order_by( DataAccessCompiledEffect.target_project_id, DataAccessCompiledEffect.target_dataset_id, DataAccessCompiledEffect.target_table_id, DataAccessCompiledEffect.id, ) .with_for_update(skip_locked=True) ) if effect_ids is not None: stmt = stmt.where(DataAccessCompiledEffect.id.in_(effect_ids)) return list((await db.scalars(stmt)).all()) async def _load_apply_gate_config( db: AsyncSession, *, integration_id: str, ) -> tuple[DataAccessApplyGateConfig, str | None]: try: integration_uuid = uuid.UUID(integration_id) except ValueError as exc: raise AppException( code="DATA_ACCESS_INTEGRATION_INVALID", message="Data-access integration id must be a UUID.", status_code=422, ) from exc integration = await db.get(Integration, integration_uuid) if integration is None: raise AppException( code="DATA_ACCESS_INTEGRATION_NOT_FOUND", message="Data-access integration not found.", status_code=404, ) if ( integration.kind != c.DOMAIN_DATA_ACCESS or integration.provider != c.PROVIDER_BIGQUERY or not integration.enabled ): return ( DataAccessApplyGateConfig( apply_mode_enabled=False, personal_apply_enabled=False, service_account_apply_enabled=False, ), "Data-access apply gate requires an enabled data_access/bigquery integration", ) try: return data_access_apply_gate_config(integration.config_public), None except ValueError as exc: return ( DataAccessApplyGateConfig( apply_mode_enabled=False, personal_apply_enabled=False, service_account_apply_enabled=False, ), str(exc), ) def _apply_gate_decision_for_effect( effect: DataAccessCompiledEffect, *, gate_config: DataAccessApplyGateConfig, config_error: str | None, ) -> DataAccessApplyGateDecision: locator = dataset_locator_key(effect.target_project_id, effect.target_dataset_id) if config_error is not None: allowed = False reason = APPLY_GATE_REASON_INVALID_CONFIG elif not gate_config.apply_mode_enabled: allowed = False reason = APPLY_GATE_REASON_DISABLED elif ( effect.desired_state == EFFECT_DESIRED_PRESENT and effect.resource is None ): allowed = False reason = APPLY_GATE_REASON_RESOURCE_UNAVAILABLE elif ( effect.desired_state == EFFECT_DESIRED_PRESENT and effect.resource is not None and not _effect_target_matches_current_resource(effect) ): allowed = False reason = APPLY_GATE_REASON_RESOURCE_LOCATOR_MISMATCH elif ( effect.desired_state == EFFECT_DESIRED_PRESENT and effect.resource is not None and effect.resource.status != RESOURCE_STATUS_ACTIVE ): allowed = False reason = APPLY_GATE_REASON_RESOURCE_INACTIVE elif ( effect.desired_state == EFFECT_DESIRED_PRESENT and effect.resource is not None and not effect.resource.requestable ): allowed = False reason = APPLY_GATE_REASON_RESOURCE_NOT_REQUESTABLE elif ( effect.desired_state == EFFECT_DESIRED_PRESENT and effect.resource is not None and not effect.resource.resource_owner_authority_enabled ): allowed = False reason = APPLY_GATE_REASON_RESOURCE_OWNER_AUTHORITY_UNAVAILABLE elif effect.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER: if not gate_config.personal_apply_enabled: allowed = False reason = APPLY_GATE_REASON_PERSONAL_DISABLED else: allowed = True reason = APPLY_GATE_REASON_ALLOWED elif effect.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT: # Typed SA requests are authorized by approval, not an operator's # pre-registration list. The compiled effect fixes the exact member # and scope; the live-resource checks above still apply to grants. if not gate_config.service_account_apply_enabled: allowed = False reason = APPLY_GATE_REASON_SERVICE_ACCOUNT_DISABLED else: allowed = True reason = APPLY_GATE_REASON_ALLOWED else: allowed = True reason = APPLY_GATE_REASON_ALLOWED return DataAccessApplyGateDecision( effect_id=effect.id, resource_id=effect.resource_id, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, provider_member=effect.provider_member, principal_id=effect.principal_id, principal_type_snapshot=effect.principal_type_snapshot, principal_subject_snapshot=effect.principal_subject_snapshot, provider_access_kind=effect.provider_access_kind, provider_role=effect.provider_role, desired_state=effect.desired_state, locator=locator, allowed=allowed, reason=reason, target_table_id=getattr(effect, "target_table_id", None), ) def _effect_target_matches_current_resource(effect: DataAccessCompiledEffect) -> bool: resource = effect.resource if resource is None: return False return ( resource.provider == effect.provider and resource.resource_type == effect.resource_type and resource.project_id == effect.target_project_id and resource.dataset_id == effect.target_dataset_id and resource.table_id == getattr(effect, "target_table_id", None) ) def _present_effects_apply_patch( effects: Sequence[DataAccessCompiledEffect], ) -> DataAccessDatasetIamPolicyPatch: members_by_role: dict[str, set[str]] = {} for effect in effects: members_by_role.setdefault(effect.provider_role, set()).add( effect.provider_member, ) return DataAccessDatasetIamPolicyPatch( add_bindings=tuple( DataAccessDatasetIamBinding( role=role, members=tuple(sorted(members)), ) for role, members in sorted(members_by_role.items()) ), remove_bindings=(), ) def _absent_effects_apply_patch( effects: Sequence[DataAccessCompiledEffect], ) -> DataAccessDatasetIamPolicyPatch: members_by_role: dict[str, set[str]] = {} for effect in effects: members_by_role.setdefault(effect.provider_role, set()).add( effect.provider_member, ) return DataAccessDatasetIamPolicyPatch( add_bindings=(), remove_bindings=tuple( DataAccessDatasetIamBinding( role=role, members=tuple(sorted(members)), ) for role, members in sorted(members_by_role.items()) ), ) def _is_revoked_binding_absent_effect(effect: DataAccessCompiledEffect) -> bool: return ( effect.binding is not None and effect.binding.status in {BINDING_STATUS_REVOKED_PENDING, BINDING_STATUS_REVOKED} ) def _is_active_binding_present_effect(effect: DataAccessCompiledEffect) -> bool: """A PRESENT effect may only be applied while its binding is ACTIVE. A binding with no row (defensive) or one already flipped to ``revoked_pending`` / ``revoked`` must not have its group re-written into the dataset IAM policy — that would resurrect an access grant that has already been revoked Helix-side. """ return ( effect.binding is not None and effect.binding.status == BINDING_STATUS_ACTIVE ) def _apply_present_result_for_update( db: AsyncSession, *, effect: DataAccessCompiledEffect, integration_id: str, run_id: uuid.UUID, trace_id: str, update: DataAccessDatasetIamUpdateResult, fallback_etag: str | None, already_present_before_apply: bool, started_at: datetime, ) -> DataAccessApplyPresentEffectResult: provider_etag = update.etag or fallback_etag if update.ok: already_applied = update.already_applied or already_present_before_apply outcome = ( APPLY_PRESENT_OUTCOME_ALREADY_APPLIED if already_applied else APPLY_PRESENT_OUTCOME_APPLIED ) if already_applied: effect.already_present_at_first_apply = True return _record_apply_present_result( db, effect=effect, integration_id=integration_id, run_id=run_id, trace_id=trace_id, outcome=outcome, status=EFFECT_STATUS_APPLIED, provider_etag=provider_etag, error_message=None, started_at=started_at, ) outcome = ( APPLY_PRESENT_OUTCOME_CONFLICT if update.conflict else APPLY_PRESENT_OUTCOME_ERROR ) return _record_apply_present_result( db, effect=effect, integration_id=integration_id, run_id=run_id, trace_id=trace_id, outcome=outcome, status=EFFECT_STATUS_RETRYABLE, provider_etag=provider_etag, error_message=update.error_message, started_at=started_at, ) def _record_apply_present_result( db: AsyncSession, *, effect: DataAccessCompiledEffect, integration_id: str, run_id: uuid.UUID, trace_id: str, outcome: str, status: str, provider_etag: str | None, error_message: str | None, started_at: datetime, ) -> DataAccessApplyPresentEffectResult: effect.status = status effect.last_provider_etag = provider_etag effect.last_provider_outcome = outcome effect.last_error = error_message _record_effect_run( db, effect=effect, integration_id=integration_id, run_id=run_id, trace_id=trace_id, mode="apply_present", outcome=outcome, provider_etag=provider_etag, error_message=error_message, started_at=started_at, ) return DataAccessApplyPresentEffectResult( effect_id=effect.id, resource_id=effect.resource_id, 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, outcome=outcome, already_present_at_first_apply=effect.already_present_at_first_apply, provider_etag=provider_etag, error_message=error_message, ) def _apply_absent_result_for_update( db: AsyncSession, *, effect: DataAccessCompiledEffect, integration_id: str, run_id: uuid.UUID, trace_id: str, update: DataAccessDatasetIamUpdateResult, fallback_etag: str | None, started_at: datetime, ) -> DataAccessApplyAbsentEffectResult: provider_etag = update.etag or fallback_etag if update.ok: outcome = ( APPLY_ABSENT_OUTCOME_ALREADY_ABSENT if update.already_applied else APPLY_ABSENT_OUTCOME_APPLIED ) return _record_apply_absent_result( db, effect=effect, integration_id=integration_id, run_id=run_id, trace_id=trace_id, outcome=outcome, status=EFFECT_STATUS_APPLIED, provider_etag=provider_etag, error_message=None, started_at=started_at, ) outcome = ( APPLY_ABSENT_OUTCOME_CONFLICT if update.conflict else APPLY_ABSENT_OUTCOME_ERROR ) return _record_apply_absent_result( db, effect=effect, integration_id=integration_id, run_id=run_id, trace_id=trace_id, outcome=outcome, status=EFFECT_STATUS_RETRYABLE, provider_etag=provider_etag, error_message=update.error_message, started_at=started_at, ) def _record_apply_absent_result( db: AsyncSession, *, effect: DataAccessCompiledEffect, integration_id: str, run_id: uuid.UUID, trace_id: str, outcome: str, status: str, provider_etag: str | None, error_message: str | None, started_at: datetime, ) -> DataAccessApplyAbsentEffectResult: effect.status = status effect.last_provider_etag = provider_etag effect.last_provider_outcome = outcome effect.last_error = error_message if ( status == EFFECT_STATUS_APPLIED and effect.binding is not None and effect.binding.status == BINDING_STATUS_REVOKED_PENDING ): effect.binding.status = BINDING_STATUS_REVOKED if effect.binding.revoked_at is None: effect.binding.revoked_at = datetime.now(UTC) _record_effect_run( db, effect=effect, integration_id=integration_id, run_id=run_id, trace_id=trace_id, mode="apply_absent", outcome=outcome, provider_etag=provider_etag, error_message=error_message, started_at=started_at, ) return DataAccessApplyAbsentEffectResult( effect_id=effect.id, resource_id=effect.resource_id, 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, outcome=outcome, provider_etag=provider_etag, error_message=error_message, ) async def _load_managed_members_by_locator( db: AsyncSession, effects: Sequence[DataAccessCompiledEffect], ) -> dict[tuple[str, str, str | None], set[tuple[str, str]]]: resource_ids = {effect.resource_id for effect in effects} locators = { ( effect.target_project_id, effect.target_dataset_id, effect.target_table_id, ) for effect in effects } if not resource_ids or not locators: return {} rows = ( await db.execute( select( DataAccessCompiledEffect.target_project_id, DataAccessCompiledEffect.target_dataset_id, DataAccessCompiledEffect.target_table_id, DataAccessCompiledEffect.provider_role, DataAccessCompiledEffect.provider_member, ).where( DataAccessCompiledEffect.resource_id.in_(resource_ids), DataAccessCompiledEffect.provider == c.PROVIDER_BIGQUERY, DataAccessCompiledEffect.resource_type.in_( (c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE), ), ), ) ).all() managed: dict[tuple[str, str, str | None], set[tuple[str, str]]] = {} for project_id, dataset_id, table_id, provider_role, provider_member in rows: locator = (project_id, dataset_id, table_id) if locator not in locators: continue if not _is_reportable_provider_grant(provider_role, provider_member): continue managed.setdefault(locator, set()).add((provider_role, provider_member)) return managed def _record_effect_run( db: AsyncSession, *, effect: DataAccessCompiledEffect, integration_id: str, run_id: uuid.UUID, trace_id: str, outcome: str, provider_etag: str | None, error_message: str | None, started_at: datetime, mode: str = "observe", ) -> None: db.add(DataAccessEffectRun( run_id=run_id, effect_id=effect.id, binding_id=effect.binding_id, request_id=( effect.binding.source_request_id if effect.binding is not None else None ), resource_id=effect.resource_id, integration_id=integration_id, trace_id=trace_id, mode=mode, 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, principal_id=effect.principal_id, principal_type_snapshot=effect.principal_type_snapshot, principal_subject_snapshot=effect.principal_subject_snapshot, provider_access_kind=effect.provider_access_kind, provider_role=effect.provider_role, desired_state=effect.desired_state, effect_status=effect.status, provider_outcome=outcome, provider_etag=provider_etag, error_message=error_message, started_at=started_at, completed_at=datetime.now(UTC), )) async def _refresh_reconcile_rollups( db: AsyncSession, *, effects: Sequence[DataAccessCompiledEffect], run_id: uuid.UUID, observed_at: datetime, ) -> None: binding_ids = {effect.binding_id for effect in effects} if not binding_ids: return bindings = ( await db.scalars( select(DataAccessBinding) .where(DataAccessBinding.id.in_(binding_ids)) .with_for_update(), ) ).all() binding_effect_rows = ( await db.execute( select( DataAccessCompiledEffect.binding_id, DataAccessCompiledEffect.status, DataAccessCompiledEffect.last_error, ).where(DataAccessCompiledEffect.binding_id.in_(binding_ids)), ) ).all() rows_by_binding: dict[uuid.UUID, list[tuple[str, str | None]]] = {} for binding_id, status, error in binding_effect_rows: rows_by_binding.setdefault(binding_id, []).append((status, error)) for binding in bindings: status, error = _rollup_effect_statuses( rows_by_binding.get(binding.id, []), ) binding.reconcile_status = status binding.last_reconcile_run_id = run_id binding.last_reconcile_at = observed_at binding.last_reconcile_error = error request_ids = { binding.source_request_id for binding in bindings if binding.source_request_id is not None } if not request_ids: return requests = ( await db.scalars( select(DataAccessRequest) .where(DataAccessRequest.id.in_(request_ids)) .with_for_update(), ) ).all() request_effect_rows = ( await db.execute( select( DataAccessBinding.source_request_id, DataAccessCompiledEffect.status, DataAccessCompiledEffect.last_error, ) .join( DataAccessBinding, DataAccessCompiledEffect.binding_id == DataAccessBinding.id, ) .where(DataAccessBinding.source_request_id.in_(request_ids)), ) ).all() rows_by_request: dict[uuid.UUID, list[tuple[str, str | None]]] = {} for request_id, status, error in request_effect_rows: if request_id is None: continue rows_by_request.setdefault(request_id, []).append((status, error)) for request in requests: status, error = _rollup_effect_statuses( rows_by_request.get(request.id, []), ) request.reconcile_status = status request.last_reconcile_run_id = run_id request.last_reconcile_at = observed_at request.last_reconcile_error = error def _rollup_effect_statuses( rows: Sequence[tuple[str, str | None]], ) -> tuple[str, str | None]: if not rows: return RECONCILE_STATUS_NOT_STARTED, None statuses = {status for status, _ in rows} error = next((err for _, err in rows if err), None) if EFFECT_STATUS_FAILED in statuses: return RECONCILE_STATUS_FAILED, error if EFFECT_STATUS_RETRYABLE in statuses: return RECONCILE_STATUS_RETRYABLE, error if EFFECT_STATUS_PENDING_RECONCILE in statuses: return RECONCILE_STATUS_PENDING, None if statuses <= {EFFECT_STATUS_APPLIED}: return RECONCILE_STATUS_APPLIED, None if statuses <= {EFFECT_STATUS_OBSERVED, EFFECT_STATUS_APPLIED}: return RECONCILE_STATUS_OBSERVED, None return RECONCILE_STATUS_PENDING, None def _trace_id_text( trace_id: str | uuid.UUID | None, integration_id: str, *, default_prefix: str = "data_access_observe", ) -> str: text = str(trace_id or f"{default_prefix}:{integration_id}").strip() return text[:128] def _unmanaged_provider_grants( locator_effects: Sequence[DataAccessCompiledEffect], *, policy: DataAccessDatasetIamPolicy, managed_provider_grants: set[tuple[str, str]], ) -> tuple[DataAccessUnmanagedProviderGrant, ...]: if not locator_effects: return () resource_id = sorted( {effect.resource_id for effect in locator_effects}, key=str, )[0] project_id = locator_effects[0].target_project_id dataset_id = locator_effects[0].target_dataset_id table_id = locator_effects[0].target_table_id unmanaged: set[tuple[str, str]] = set() for binding in policy.bindings: for provider_member in binding.members: if not _is_reportable_provider_grant(binding.role, provider_member): continue provider_grant = (binding.role, provider_member) if provider_grant not in managed_provider_grants: unmanaged.add(provider_grant) return tuple( DataAccessUnmanagedProviderGrant( resource_id=resource_id, target_project_id=project_id, target_dataset_id=dataset_id, target_table_id=table_id, provider_member=provider_member, provider_role=provider_role, provider_etag=policy.etag, ) for provider_role, provider_member in sorted( unmanaged, key=lambda grant: (grant[0], grant[1]), ) ) def _is_reportable_provider_grant(provider_role: str, provider_member: str) -> bool: return ( provider_role == c.BIGQUERY_DATA_VIEWER_ROLE and provider_member.startswith(( c.BIGQUERY_GROUP_MEMBER_PREFIX, c.BIGQUERY_SERVICE_ACCOUNT_MEMBER_PREFIX, c.BIGQUERY_USER_MEMBER_PREFIX, )) ) def _group_effects_by_locator( effects: Sequence[DataAccessCompiledEffect], ) -> dict[tuple[str, str, str | None], list[DataAccessCompiledEffect]]: grouped: dict[tuple[str, str, str | None], list[DataAccessCompiledEffect]] = {} for effect in effects: grouped.setdefault( ( effect.target_project_id, effect.target_dataset_id, effect.target_table_id, ), [], ).append(effect) return grouped async def _get_locator_iam( driver: DataAccessDriver, project_id: str, dataset_id: str, table_id: str | None, ) -> DataAccessDatasetIamPolicy: if table_id: return await driver.get_table_iam(project_id, dataset_id, table_id) return await driver.get_dataset_iam(project_id, dataset_id) async def _set_locator_iam( driver: DataAccessDriver, project_id: str, dataset_id: str, table_id: str | None, expected_etag: str | None, patch: DataAccessDatasetIamPolicyPatch, ) -> DataAccessDatasetIamUpdateResult: if table_id: return await driver.set_table_iam( project_id, dataset_id, table_id, expected_etag, patch, ) return await driver.set_dataset_iam( project_id, dataset_id, expected_etag, patch, ) def _members_by_role( policy: DataAccessDatasetIamPolicy, ) -> dict[str, set[str]]: grouped: dict[str, set[str]] = {} for binding in policy.bindings: grouped.setdefault(binding.role, set()).update(binding.members) return grouped def _provider_has_effect( provider_members: dict[str, set[str]], effect: DataAccessCompiledEffect, ) -> bool: return effect.provider_member in provider_members.get(effect.provider_role, set()) def _observe_outcome( effect: DataAccessCompiledEffect, provider_members: dict[str, set[str]], ) -> str: provider_has_member = _provider_has_effect(provider_members, effect) if effect.desired_state == EFFECT_DESIRED_PRESENT: return ( OBSERVE_OUTCOME_UNCHANGED if provider_has_member else OBSERVE_OUTCOME_WOULD_ADD ) if effect.desired_state == EFFECT_DESIRED_ABSENT: return ( OBSERVE_OUTCOME_WOULD_REMOVE if provider_has_member else OBSERVE_OUTCOME_UNCHANGED ) return OBSERVE_OUTCOME_ERROR def _delta_for_effect( effect: DataAccessCompiledEffect, *, outcome: str, provider_etag: str | None, error_message: str | None = None, ) -> DataAccessObserveEffectDelta: return DataAccessObserveEffectDelta( effect_id=effect.id, resource_id=effect.resource_id, 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, outcome=outcome, provider_etag=provider_etag, error_message=error_message, ) async def _audit_observe_result( db: AsyncSession, result: DataAccessObserveReconcileResult, *, trace_id: str | uuid.UUID | None, run_id: uuid.UUID, effect_ids: tuple[uuid.UUID, ...] | None, resource_ids: tuple[uuid.UUID, ...] | None, ) -> None: await AuditService(db).log( trace_id=trace_id or f"data_access_observe:{result.integration_id}", actor_type="integration", action="data_access.reconcile.observe", status="failure" if result.error_count else "success", resource_type="data_access_integration", resource_id=result.integration_id, detail={ "mode": "observe", "run_id": str(run_id), "lock_strategy": "for_update_skip_locked", "provider_mutation_enabled": False, "selection": { "effect_ids": ( None if effect_ids is None else [str(effect_id) for effect_id in effect_ids] ), "resource_ids": ( None if resource_ids is None else [str(resource_id) for resource_id in resource_ids] ), }, "observed_effect_count": result.observed_effect_count, "would_add_count": result.would_add_count, "would_remove_count": result.would_remove_count, "unchanged_count": result.unchanged_count, "error_count": result.error_count, "unmanaged_grant_count": result.unmanaged_grant_count, "deltas": [ { "effect_id": str(delta.effect_id), "resource_id": str(delta.resource_id), "target_project_id": delta.target_project_id, "target_dataset_id": delta.target_dataset_id, "provider_member": delta.provider_member, "provider_role": delta.provider_role, "desired_state": delta.desired_state, "outcome": delta.outcome, "provider_etag": delta.provider_etag, "error_message": delta.error_message, } for delta in result.deltas ], "unmanaged_grants": [ { "resource_id": str(grant.resource_id), "target_project_id": grant.target_project_id, "target_dataset_id": grant.target_dataset_id, "provider_member": grant.provider_member, "provider_role": grant.provider_role, "provider_etag": grant.provider_etag, } for grant in result.unmanaged_grants ], }, ) async def _audit_apply_gate_result( db: AsyncSession, result: DataAccessApplyGateResult, *, trace_id: str | uuid.UUID | None, run_id: uuid.UUID, effect_ids: tuple[uuid.UUID, ...] | None, resource_ids: tuple[uuid.UUID, ...] | None, ) -> None: await AuditService(db).log( trace_id=trace_id or f"data_access_apply_gate:{result.integration_id}", actor_type="integration", action="data_access.reconcile.apply_gate", status="failure" if result.config_error else "success", resource_type="data_access_integration", resource_id=result.integration_id, detail={ "mode": "apply_gate", "run_id": str(run_id), "lock_strategy": "for_update_skip_locked", "provider_mutation_attempted": False, "apply_mode_enabled": result.apply_mode_enabled, "config_error": result.config_error, "selection": { "effect_ids": ( None if effect_ids is None else [str(effect_id) for effect_id in effect_ids] ), "resource_ids": ( None if resource_ids is None else [str(resource_id) for resource_id in resource_ids] ), }, "selected_effect_count": result.selected_effect_count, "allowed_effect_count": result.allowed_effect_count, "blocked_effect_count": result.blocked_effect_count, "decisions": [ { "effect_id": str(decision.effect_id), "resource_id": str(decision.resource_id), "target_project_id": decision.target_project_id, "target_dataset_id": decision.target_dataset_id, "provider_member": decision.provider_member, "provider_role": decision.provider_role, "desired_state": decision.desired_state, "locator": decision.locator, "allowed": decision.allowed, "reason": decision.reason, } for decision in result.decisions ], }, ) async def _audit_apply_present_result( db: AsyncSession, result: DataAccessApplyPresentReconcileResult, *, trace_id: str | uuid.UUID | None, run_id: uuid.UUID, effect_ids: tuple[uuid.UUID, ...] | None, resource_ids: tuple[uuid.UUID, ...] | None, ) -> None: await AuditService(db).log( trace_id=trace_id or f"data_access_apply_present:{result.integration_id}", actor_type="integration", action="data_access.reconcile.apply_present", status=( "failure" if result.config_error or result.conflict_count or result.error_count else "success" ), resource_type="data_access_integration", resource_id=result.integration_id, detail={ "mode": "apply_present", "run_id": str(run_id), "lock_strategy": "for_update_skip_locked", "provider_mutation_attempted": result.provider_write_count > 0, "provider_write_count": result.provider_write_count, "apply_mode_enabled": result.apply_mode_enabled, "config_error": result.config_error, "selection": { "effect_ids": ( None if effect_ids is None else [str(effect_id) for effect_id in effect_ids] ), "resource_ids": ( None if resource_ids is None else [str(resource_id) for resource_id in resource_ids] ), }, "selected_effect_count": result.selected_effect_count, "selected_present_effect_count": result.selected_present_effect_count, "ignored_non_present_effect_count": ( result.ignored_non_present_effect_count ), "allowed_effect_count": result.allowed_effect_count, "blocked_effect_count": result.blocked_effect_count, "applied_count": result.applied_count, "already_applied_count": result.already_applied_count, "conflict_count": result.conflict_count, "error_count": result.error_count, "gate_decisions": [ { "effect_id": str(decision.effect_id), "resource_id": str(decision.resource_id), "target_project_id": decision.target_project_id, "target_dataset_id": decision.target_dataset_id, "provider_member": decision.provider_member, "provider_role": decision.provider_role, "desired_state": decision.desired_state, "locator": decision.locator, "allowed": decision.allowed, "reason": decision.reason, } for decision in result.gate_decisions ], "results": [ { "effect_id": str(row.effect_id), "resource_id": str(row.resource_id), "target_project_id": row.target_project_id, "target_dataset_id": row.target_dataset_id, "provider_member": row.provider_member, "provider_role": row.provider_role, "desired_state": row.desired_state, "outcome": row.outcome, "already_present_at_first_apply": ( row.already_present_at_first_apply ), "provider_etag": row.provider_etag, "error_message": row.error_message, } for row in result.results ], }, ) async def _audit_apply_absent_result( db: AsyncSession, result: DataAccessApplyAbsentReconcileResult, *, trace_id: str | uuid.UUID | None, run_id: uuid.UUID, effect_ids: tuple[uuid.UUID, ...] | None, resource_ids: tuple[uuid.UUID, ...] | None, ) -> None: await AuditService(db).log( trace_id=trace_id or f"data_access_apply_absent:{result.integration_id}", actor_type="integration", action="data_access.reconcile.apply_absent", status=( "failure" if result.config_error or result.conflict_count or result.error_count else "success" ), resource_type="data_access_integration", resource_id=result.integration_id, detail={ "mode": "apply_absent", "run_id": str(run_id), "lock_strategy": "for_update_skip_locked", "provider_mutation_attempted": result.provider_write_count > 0, "provider_write_count": result.provider_write_count, "apply_mode_enabled": result.apply_mode_enabled, "config_error": result.config_error, "selection": { "effect_ids": ( None if effect_ids is None else [str(effect_id) for effect_id in effect_ids] ), "resource_ids": ( None if resource_ids is None else [str(resource_id) for resource_id in resource_ids] ), }, "selected_effect_count": result.selected_effect_count, "selected_absent_effect_count": result.selected_absent_effect_count, "ignored_non_absent_effect_count": ( result.ignored_non_absent_effect_count ), "ignored_non_revoked_binding_effect_count": ( result.ignored_non_revoked_binding_effect_count ), "allowed_effect_count": result.allowed_effect_count, "blocked_effect_count": result.blocked_effect_count, "removed_count": result.removed_count, "already_absent_count": result.already_absent_count, "conflict_count": result.conflict_count, "error_count": result.error_count, "gate_decisions": [ { "effect_id": str(decision.effect_id), "resource_id": str(decision.resource_id), "target_project_id": decision.target_project_id, "target_dataset_id": decision.target_dataset_id, "provider_member": decision.provider_member, "provider_role": decision.provider_role, "desired_state": decision.desired_state, "locator": decision.locator, "allowed": decision.allowed, "reason": decision.reason, } for decision in result.gate_decisions ], "results": [ { "effect_id": str(row.effect_id), "resource_id": str(row.resource_id), "target_project_id": row.target_project_id, "target_dataset_id": row.target_dataset_id, "provider_member": row.provider_member, "provider_role": row.provider_role, "desired_state": row.desired_state, "outcome": row.outcome, "provider_etag": row.provider_etag, "error_message": row.error_message, } for row in result.results ], }, ) def _truncate_error(exc: BaseException) -> str: msg = _PRIVATE_KEY_RE.sub("", str(exc)) return msg[:500] __all__ = [ "APPLY_ABSENT_OUTCOME_ALREADY_ABSENT", "APPLY_ABSENT_OUTCOME_APPLIED", "APPLY_ABSENT_OUTCOME_CONFLICT", "APPLY_ABSENT_OUTCOME_ERROR", "APPLY_GATE_REASON_ALLOWED", "APPLY_GATE_REASON_DISABLED", "APPLY_GATE_REASON_INVALID_CONFIG", "APPLY_PRESENT_OUTCOME_ALREADY_APPLIED", "APPLY_PRESENT_OUTCOME_APPLIED", "APPLY_PRESENT_OUTCOME_CONFLICT", "APPLY_PRESENT_OUTCOME_ERROR", "DataAccessApplyAbsentEffectResult", "DataAccessApplyAbsentReconcileResult", "DataAccessApplyGateDecision", "DataAccessApplyGateResult", "DataAccessApplyPresentEffectResult", "DataAccessApplyPresentReconcileResult", "DataAccessObserveEffectDelta", "DataAccessObserveReconcileResult", "DataAccessUnmanagedProviderGrant", "OBSERVE_OUTCOME_ERROR", "OBSERVE_OUTCOME_UNCHANGED", "OBSERVE_OUTCOME_WOULD_ADD", "OBSERVE_OUTCOME_WOULD_REMOVE", "apply_absent_reconcile_compiled_effects", "apply_present_reconcile_compiled_effects", "observe_reconcile_compiled_effects", "resolve_apply_reconcile_gate", ]