"""Google Workspace custom-schema field runner. Pushes rendered fields into Google Directory's ``users.customSchemas.`` namespace via ``users.patch``. This runner replaces the legacy ``google.dept_sync`` ScheduledJob — the per-target ``runner_options`` carry the schema_id and root-segment strip list that used to live in ``platform_settings.google.dept_schema``. Why a separate runner instead of squeezing into SCIM: * Google's Admin Directory API isn't SCIM. It has its own auth model (service-account + domain-wide delegation), its own field shape (``customSchemas`` is a nested dict per schema), and it doesn't support SCIM-style ``active`` toggling for our use case. * The dept-fields use case is "write a few attributes on existing users" — not "create / deactivate users". The runner's ``deactivate`` is a no-op (we don't manage Google account lifecycle here — that lives in ``google_workspace`` domain's account_provisioner) and ``upsert`` only touches the configured schema. Configuration (target.runner_options): { "schema_id": "Department_Information", "drop_root_segments": ["PingCAP"] } The mapping rules use ``{org_path_level_N}`` source templates and their target_field is the bare schema field name (e.g. ``L1_Dept``). The runner combines the schema_id + the rendered fields into the nested customSchemas body that ``patch_user_department`` expects. """ from __future__ import annotations import logging from typing import Any from app.domains.identity_sync.runners.base import PersonOutcome, SyncRunner from app.integrations.drivers.google_workspace import ( GoogleApiError, GoogleWorkspaceDriver, ) logger = logging.getLogger(__name__) class GoogleSchemaRunner(SyncRunner): """Custom-schema field projection on Google Workspace users. Google account lifecycle (suspend / unsuspend / delete) is owned by ``google_workspace.account_provisioner`` — the Identity Sync Center never crosses that boundary. So: * ``supports_deactivate = False`` — when admin's on_offboard is 'deactivate' on a google_schema target, the orchestrator marks the state row 'excluded' without calling Google. That matches semantic intent: "stop projecting fields onto this person's record" rather than "suspend the Google account." * ``supports_delete = True`` — exposed because ``delete`` here means clearing the configured customSchemas fields (real, reversible work that doesn't touch account state). """ supports_deactivate = False supports_delete = True def __init__( self, *, driver: GoogleWorkspaceDriver, schema_id: str, ): if not schema_id: raise ValueError("google_schema runner requires schema_id") self._driver = driver self._schema_id = schema_id # Cached schema validation result. We check schema existence # once per runner lifetime — populated on first use. self._schema_validated: bool | None = None # Cached "schema missing" message so a FALSE result re-raises # without re-querying (see _ensure_schema_exists). self._schema_error: str | None = None async def _ensure_schema_exists(self) -> None: """Bail loudly when the configured schema isn't in the tenant. Without this guard a misconfigured schema_id would generate one 404 per person — burning quota and producing a result page full of identical errors. Validate once, raise if missing, let the run wrap that into a single error_message. Both outcomes are CACHED: a prior success short-circuits, and a prior MISS re-raises the cached error WITHOUT re-querying ``list_custom_schemas`` — otherwise every subsequent person in the sweep re-ran the full schema list call (an N-call quota storm), the exact thing this guard exists to prevent. """ if self._schema_validated is True: return if self._schema_validated is False: raise GoogleApiError( self._schema_error or f"schema {self._schema_id!r} not found in tenant; " "fix runner_options.schema_id.", ) schemas = await self._driver.list_custom_schemas() names = [s.get("schemaName") for s in schemas] if self._schema_id not in names: self._schema_validated = False self._schema_error = ( f"schema {self._schema_id!r} not found in tenant " f"(available: {names}); fix runner_options.schema_id." ) raise GoogleApiError(self._schema_error) self._schema_validated = True async def upsert( self, *, person_email: str, payload: dict[str, Any], prior_external_id: str | None, ) -> PersonOutcome: """Patch the user's customSchemas slot. ``payload`` is whatever the mapping engine produced — for the Google use case that's a flat dict of schema field names → values (e.g. ``{"L1_Dept": "Eng", "L2_Dept": "Infra"}``). We wrap it as ``{customSchemas: {: }}`` and send through the existing ``patch_user_department`` helper. Missing target_field renders are dropped by the mapping engine (we get the value as absent in payload), but the legacy job's behavior was to send ``null`` to clear stale Google-side values. Preserve that: any field in the schema that ISN'T in the payload becomes ``null`` in the patch body, but ONLY for the four conventional level fields; if the admin extends the schema with extra fields they manage themselves, leave them alone. """ await self._ensure_schema_exists() # No Google account yet — get_user_profile returns None. This is a # TEMPORARY state: onboarding provisioning often lands the account # AFTER this schema sweep runs (build lead-window not reached, or a # partial create). Returning ``skipped`` here would let the # orchestrator store ``last_payload_hash`` and treat this person as # permanently unchanged; once the account appears the payload is # unchanged, so every future sweep hash-skips and the # customSchemas department fields are NEVER written (#1759). # ``skipped`` semantics fit a PERMANENT refusal (e.g. SCIM), not a # "not there yet" — return ``failed`` so the sweep retries until # the account exists and the fields get pushed. profile = await self._driver.get_user_profile(person_email) if profile is None: return PersonOutcome( status="failed", error=( "no Google account for this email yet — retrying next " "sweep until provisioning creates it" ), ) # Build the field set from the rendered payload directly. The # mapping engine drops absent rules, so a 2-level dept won't # render L3 / L4 here. To clear stale Google-side values for # those levels, admins must add explicit mapping rules with a # ``default: ""`` — the renderer then emits the empty string # and Google interprets it as "clear." We deliberately do NOT # auto-pad missing slots with ``None`` here because that # would silently overwrite fields the admin chose not to map # (e.g. an extension schema field they manage out-of-band). fields: dict[str, str | None] = dict(payload) # Diff: skip if customSchemas already matches. existing = (profile.get("customSchemas") or {}).get(self._schema_id) or {} if all((existing.get(k) or None) == v for k, v in fields.items()): return PersonOutcome(status="skipped") result = await self._driver.patch_user_department( user_email=person_email, schema_id=self._schema_id, fields=fields, ) if result.ok: return PersonOutcome(status="success", external_id=person_email) return PersonOutcome( status="failed", error=result.error_message or "unknown", ) # ``deactivate`` intentionally NOT implemented — see class # docstring. ``supports_deactivate=False`` tells the orchestrator # to short-circuit to ``status='excluded'`` without calling here. # Calling it directly raises NotImplementedError from the base. async def delete( self, *, person_email: str, prior_external_id: str | None, ) -> PersonOutcome: """Clear the configured schema fields from the Google user. Sends each level slot as ``None`` in the patch body so Google wipes the stored value. The user record itself is not deleted. """ # Best-effort schema validation — if the schema is gone we # still treat the delete as success (nothing to clear). try: await self._ensure_schema_exists() except GoogleApiError: return PersonOutcome(status="deleted") profile = await self._driver.get_user_profile(person_email) if profile is None: return PersonOutcome(status="deleted") existing = (profile.get("customSchemas") or {}).get(self._schema_id) or {} if not existing: return PersonOutcome(status="deleted") result = await self._driver.patch_user_department( user_email=person_email, schema_id=self._schema_id, fields={k: None for k in existing.keys()}, ) if result.ok: return PersonOutcome(status="deleted") return PersonOutcome( status="failed", error=result.error_message or "unknown", ) def describe(self) -> str: return f"Google customSchemas/{self._schema_id}"