"""Seed google_schema target for the Employee_Details customSchema. Revision ID: 363 Revises: 362 Create Date: 2026-06-29 What this migration does ------------------------ Idempotent seed of a second ``google_schema`` identity-sync target — sibling to the live ``Google Workspace · 部门字段`` target that already projects ``Department_Information.L1..L4``. This one projects the ``Employee_Details`` customSchema. Mapping is read from the ``google_employee_details_schema`` preset and projects five SF-verbatim fields (no Helix intermediate value-map): the sources are the SuccessFactors-translated labels resolved in the HRIS sync, written straight through — * ``Office`` ← ``{office_mail_group}`` (e.g. ``sh-office@pingcap.com``) * ``JobFunction`` ← ``{job_function}`` * ``JobFamily`` ← ``{job_family}`` * ``Location`` ← ``{location}`` (city) * ``Country`` ← ``{residence_country}`` (ISO-2) ~40 Cloud Identity dynamic groups match on these raw SF strings, so no intermediate mapping table is required. (The earlier PR1 design that populated ``Office`` ONLY via the ``OFFICE_CODE_TO_FEISHU_LOCATION`` value-map was superseded in the same branch — see the live preset.) Reuses the existing google_workspace ``software_provisioning`` integration (same SA / domain-wide delegation that backs the dept target + the account provisioner). No new integration row, no new credentials. Seeded ``enabled=0`` — it goes live together with the dynamic-group query rewrite, opened via the admin UI / rollout step (not auto-armed by the migration). If the integration can't be resolved (fresh dev DB without Google configured) the row is still seeded ``enabled=0`` so a credential-less target never fires. Re-running on an already-seeded DB is a no-op (slug-uniqueness probe). """ from __future__ import annotations import json import logging import uuid import sqlalchemy as sa from alembic import op logger = logging.getLogger(__name__) revision = "363" down_revision = "362" branch_labels = None depends_on = None _TARGET_SLUG = "google-employee-details" def _row_id(bind, table: str, where: str, params: dict) -> str | None: row = bind.execute( sa.text(f"SELECT id FROM {table} WHERE {where} LIMIT 1"), params, ).fetchone() return row[0] if row else None def _resolve_google_integration(bind) -> str | None: """Find the primary enabled google_workspace software_provisioning integration — mirrors ``_load_google_driver``'s lookup.""" row = bind.execute( sa.text( "SELECT id FROM integrations " "WHERE kind = 'software_provisioning' " " AND provider = 'google_workspace' AND enabled = 1 " "ORDER BY is_primary DESC, created_at LIMIT 1" ) ).fetchone() return row[0] if row else None def _preset_seed() -> dict: """Read the live preset so the seeded mapping_config matches it.""" try: from app.domains.identity_sync.presets import get_preset p = get_preset("google_employee_details_schema") if p is None: raise RuntimeError("preset google_employee_details_schema not found") return p.to_seed() except Exception as exc: # noqa: BLE001 — migration MUST NOT raise logger.warning("migration 363: preset load failed (%r); using inline fallback", exc) return { "kind": "google_schema", "mapping_config": {"rules": [ # SF-verbatim — no value_map. Sources are the SF-translated # labels resolved in the SuccessFactors sync. {"source_template": "{office_mail_group}", "target_field": "Office"}, {"source_template": "{job_function}", "target_field": "JobFunction"}, {"source_template": "{job_family}", "target_field": "JobFamily"}, {"source_template": "{location}", "target_field": "Location"}, {"source_template": "{residence_country}", "target_field": "Country"}, ]}, "filter_config": { "include_offboarded": False, "departments": [], "include_subtree": True, "countries": [], "employment_statuses": ["active", "on_leave"], "job_title_includes": [], "person_ids": [], "mode": "and", }, "schedule_config": { "trigger": "cron", "cron_expr": "0 1 * * *", "timezone": "UTC", }, "runner_options": {"schema_id": "Employee_Details"}, } def upgrade() -> None: bind = op.get_bind() if _row_id(bind, "identity_sync_targets", "slug = :s", {"s": _TARGET_SLUG}): return # already seeded integ_id = _resolve_google_integration(bind) seed = _preset_seed() # Always seed enabled=0. This target writes fields that ~40 Cloud # Identity dynamic groups key on (Office/JobFunction/Location/Country); # auto-firing on deploy would change group values before the group # queries are cut over to the SF-verbatim strings. The rollout enables # it deliberately, alongside the group-query rewrite, then verifies. enabled = 0 if integ_id is None: logger.warning( "migration 363: no enabled google_workspace software_provisioning " "integration found — seeding Employee_Details target with " "NULL integration; configure + enable via admin UI / rollout." ) bind.execute( sa.text( "INSERT INTO identity_sync_targets " "(id, name, slug, description, kind, integration_id, " " enabled, on_offboard, " " mapping_config, filter_config, schedule_config, runner_options, " " created_at, updated_at) " "VALUES " "(:id, :name, :slug, :desc, 'google_schema', :integ, " " :enabled, 'deactivate', " " :mapping, :filter, :schedule, :runner_opts, NOW(), NOW())" ), { "id": str(uuid.uuid4()), "name": "Google Workspace · 员工字段", "slug": _TARGET_SLUG, "desc": ( "把 HRIS 字段按 SF 原样翻译串投影成 Google customSchemas " "Employee_Details(Office=EmailGroup 'sh-office@pingcap.com'、" "JobFunction=FOJobFunction 'Engineer'、JobFamily、Location=城市、" "Country=居住国 ISO-2),供 ~40 个动态组按这些 SF 串匹配。" "无 Helix 中间映射表。seed enabled=0,随动态组 query 改写一起上线。" ), "integ": integ_id, "enabled": enabled, "mapping": json.dumps(seed["mapping_config"], ensure_ascii=False), "filter": json.dumps(seed["filter_config"], ensure_ascii=False), "schedule": json.dumps(seed["schedule_config"], ensure_ascii=False), "runner_opts": json.dumps(seed["runner_options"], ensure_ascii=False), }, ) def downgrade() -> None: bind = op.get_bind() bind.execute( sa.text("DELETE FROM identity_sync_targets WHERE slug = :s"), {"s": _TARGET_SLUG}, )