"""Identity domain models: Person, OrgMembership, ExternalIdentity. These replace the old auth_users table with a SuccessFactors-ready identity model split into three concerns: - IdentityPerson: stable identity + employment status - IdentityOrgMembership: person-to-org-node mapping (denormalized) - IdentityExternalIdentity: login + external system account mapping """ import uuid from datetime import date as date_type from datetime import datetime from decimal import Decimal from sqlalchemy import ( JSON, Boolean, Date, DateTime, ForeignKey, Index, Numeric, String, Text, UniqueConstraint, ) from sqlalchemy.orm import Mapped, mapped_column, relationship, validates from sqlalchemy.sql import func from app.core.database import Base from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role, normalize_role_names from app.core.sql_types import GUID as Uuid class IdentityPerson(Base): __tablename__ = "identity_persons" __table_args__ = ( UniqueConstraint("source_system", "hris_person_id", name="uq_identity_persons_source"), UniqueConstraint("work_email", name="uq_identity_persons_work_email"), # employee_no is nullable; MySQL UNIQUE allows multiple NULLs so we use a # partial-index workaround via migration rather than a constraint here. Index("idx_identity_persons_manager", "manager_person_id"), Index("idx_identity_persons_status", "employment_status"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) # HRIS source tracking source_system: Mapped[str] = mapped_column(String(32), nullable=False, default="successfactors") # When multiple HRIS instances are configured (e.g. CN + EU SF for # data-residency compliance), this FK ties the person to the source # instance row in ``integrations``. NULL for legacy/single-instance # rows imported before multi-instance was introduced. # Declared as a plain Uuid column — the FK constraint is added at # the migration layer (036) with a named ALTER so we stay out of # the ``identity_persons`` ↔ ``integrations`` metadata cycle # (integrations carries ``created_by``/``updated_by`` FKs back to # identity_persons) that would otherwise break # ``Base.metadata.drop_all`` in unit tests. hris_instance_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) hris_person_id: Mapped[str] = mapped_column(String(128), nullable=False) employee_no: Mapped[str | None] = mapped_column(String(64)) # Core identity work_email: Mapped[str] = mapped_column(String(255), nullable=False) display_name: Mapped[str | None] = mapped_column(String(128)) name_pinyin: Mapped[str | None] = mapped_column(String(256)) name_pinyin_abbr: Mapped[str | None] = mapped_column(String(64)) # SF ``User.firstName`` (given) + ``User.lastName`` (family). Stored # raw — the value comes through whatever script HR types in (CJK # chars for Chinese employees, Latin for everyone else). Used by # the Google account provisioner to build the email local-part in # the unambiguous ``given.family`` order without having to guess # which pinyin token is the family name. See migration 240. first_name: Mapped[str | None] = mapped_column(String(128)) last_name: Mapped[str | None] = mapped_column(String(128)) # HR-assigned corporate mailbox local part ("邮箱名", e.g. ``nick.muniz``). # Authoritative source: SF ``PerPersonal.customString4`` (the same value # SF's own outbound Integration Center mapping ships as ``emailName``); # ``User.custom07`` is a legacy secondary. Migration 582. # # Why this is a COLUMN and not just an event field (P10026, 2026-08-31): # the value used to live only on the inbound event payload, read once by # the ``action in {create, reactivate}`` dispatch gate and then dropped. # An email name HR fills in AFTER the create event therefore reached # Helix on every 4-hourly sync and was discarded every time — the hire # got a ``given.family`` mailbox on day-1 even though the correct name # had been sitting in SF for nine days. Persisting it makes the mailbox # local part a function of the LATEST authoritative input at the moment # the account is actually minted, not of whatever was known the first # time Helix saw the person. # # Written apply-only by ``HrSyncService._upsert_person`` on EVERY apply # (including onboard ``skip``, which flows through ``_apply_event`` as a # refresh-only pass): a non-empty value overwrites, an absent/blank one # never clears. Blank means "HR gave no instruction", NOT "switch back # to the pinyin derivation" — clearing on absence would let one sync tick # with a partial SF read silently re-derive somebody's address. hris_email_name: Mapped[str | None] = mapped_column(String(190)) avatar_url: Mapped[str | None] = mapped_column(String(512)) # Employment employment_status: Mapped[str] = mapped_column(String(16), nullable=False, default="active") is_china_employee: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) # HRIS-sourced title (e.g. "Senior Sales Manager"). Used as one of the # three match keys for onboarding provisioning rules. job_title: Mapped[str | None] = mapped_column(String(128)) # Manager (self-referential) manager_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id") ) # HRIS-declared manager external id (SF EmpJob.managerId), persisted # verbatim so a late-arriving manager can be re-linked by # ``identity.identity_reconcile.backfill_manager_links`` long after # the subordinate's last sync event. The per-event resolvers clear # ``manager_person_id`` on a miss (a stale pointer is worse than a # brief gap — Round-20 audit); keeping the raw id here is what lets # the gap close on the next sync instead of staying NULL forever. # Also lets the data-quality scanner tell a genuine HRIS "no manager" # (NULL here → org-apex / data gap) apart from a Helix link failure # (non-NULL here, NULL ``manager_person_id`` → resolvable). Added # migration 270. manager_external_id: Mapped[str | None] = mapped_column(String(128)) # HR Business Partner (self-referential, migration 111). Sourced from # Feishu's per-user ``custom_attrs`` (the HRBP-typed entry) and # cached here so SLA escalations don't have to round-trip Feishu # twice on every dispatch. Re-resolved on next dispatch when NULL # or when the cached row is stale (resolver overwrites). NULL is a # legitimate state for system accounts / contractors / freshly- # imported persons whose Feishu record hasn't been queried yet. hrbp_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), ) # HRIS work-relationship roles (migration 291) — populated by the # SuccessFactors sync from the ``EmpJobRelationships`` (工作关系) entity, # distinct from ``hrbp_person_id`` above (which the leaver-SLA feature # resolves lazily from Feishu). Kept separate so the SF-authoritative # values feed the 汇联易 dept sync (department roles HRBP / Budget Owner) # without disturbing the leaver path. Plain Uuid columns — NO FK # constraint (self-referential CHAR(36) FKs trip TiDB 3780; the value is # a resolved person id used in code, no ORM relationship needed). # hris_hrbp_person_id ← SF relationshipType 2448 (HRBP) # hris_budget_owner_person_id ← SF relationshipType 2447 (调薪负责人) # hris_hrops_person_id ← SF relationshipType 2450 (HR Ops) hris_hrbp_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) hris_budget_owner_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) hris_hrops_person_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) # Location & entity work_location_code: Mapped[str | None] = mapped_column(String(64)) work_location_name: Mapped[str | None] = mapped_column(String(128)) # Resolved via FoLocation master data lookup (migration 048) — country # is typically a 2-letter ISO-3166 code (e.g. "CN", "US"), city is the # city name as configured in the SF location record. country: Mapped[str | None] = mapped_column(String(64)) city: Mapped[str | None] = mapped_column(String(128)) company_code: Mapped[str | None] = mapped_column(String(64)) company_name: Mapped[str | None] = mapped_column(String(128)) # SF EmpJob.costCenter — populated by the SuccessFactors sync. Used # by the Identity Sync Center to project into Navan / SCIM # enterprise.costCenter and similar downstream fields. cost_center: Mapped[str | None] = mapped_column(String(64)) # ── International assignment / 外派 (migration 392) ──────────────── # PingCAP models global assignments as custom fields on SF EmpJob # (field IDs confirmed via live $metadata labels 2026-07-07). A person # is "on international assignment" when a host country is set and differs # from the home country (e.g. 赵慧 Zita Zhao: home CHN, host SGP). These # are SUPPLEMENTARY context (surfaced in HR reports / person detail) — they # do NOT repoint the person's directory placement (office / company / dept # stay as EmpJob's home job). ISO-3 country codes stored raw (meaningful # as-is); dates parsed from SF ``/Date(ms)/``. # assignment_home_country ← EmpJob.customString44 (外派前国家) # assignment_host_country ← EmpJob.customString38 (外派后国家) # assignment_start_date ← EmpJob.customDate2 (外派开始日期) # assignment_end_date ← EmpJob.customDate4 (外派结束日期) # assignment_status_code ← EmpJob.customString39 (外派状态, RAW picklist # code — the SF status picklist is not cleanly resolvable over OData, # so the RAW code is stored for fidelity and display derives the # lifecycle from host≠home + the dates, mirroring sf_downstream_push_code) assignment_home_country: Mapped[str | None] = mapped_column(String(8)) assignment_host_country: Mapped[str | None] = mapped_column(String(8)) assignment_start_date: Mapped[date_type | None] = mapped_column(Date) assignment_end_date: Mapped[date_type | None] = mapped_column(Date) assignment_status_code: Mapped[str | None] = mapped_column(String(16)) @property def is_on_international_assignment(self) -> bool: """True when the person is CURRENTLY on an international assignment: a host country set and different from the home country, AND the assignment window hasn't ended. Picklist-free (doesn't depend on the un-resolvable 外派状态 code). The end-date guard stops an ENDED assignment (SF may retain the historical host country on the current job) from showing as active; an open-ended assignment (no end date) stays active. """ host = (self.assignment_host_country or "").strip().upper() home = (self.assignment_home_country or "").strip().upper() if not host or host == home: return False end = self.assignment_end_date return end is None or end >= date_type.today() # PII contact fields (migration 048). The UI masks both by default; # reveal is gated behind POST /api/v1/users/{id}/reveal-contact which # writes an audit_logs row naming the reader. mobile_phone: Mapped[str | None] = mapped_column(String(32)) # ISO-3166 alpha-2 of the MOBILE number itself (migration 391), synced # from SF ``PerPhone.countryCode`` (a ``CountryCode`` picklist option whose # ``externalCode`` is the ISO-2, e.g. option 1734 → ``SG``). Distinct from # ``country`` (the person's home/base country): an international-assignment # employee (home CN, on assignment to SG) carries a host-country phone # (+65), so deriving the dialing code from ``country`` (CN → +86) yields an # invalid number the vendor rejects. Downstream phone formatting (Feishu # ``mobile``) uses THIS as the dialing-code source when present, falling # back to ``country`` only when absent. mobile_country_code: Mapped[str | None] = mapped_column(String(8)) personal_email: Mapped[str | None] = mapped_column(String(255)) # User preferences synced from HRIS (migration 048). Format mirrors # SF: ``{lang}_{REGION}`` lowercase (e.g. ``en_US``, ``zh_CN``). preferred_language: Mapped[str | None] = mapped_column(String(16)) # ── HRIS attributes projected to Feishu (migration 344) ────────── # These three are stored as the RAW SF picklist code; the human / # vendor-enum translation lives in the ``feishu_contact`` identity- # sync target's value-maps (golden record stays source-faithful, # same convention as ``country`` holding the ISO-3 code). # # 性别 — SF ``PerPersonal.gender`` (``M`` / ``F``). HRIS-authoritative; # the sync overwrites when SF carries a value. → Feishu ``gender`` # enum (1=M / 2=F). gender: Mapped[str | None] = mapped_column(String(16)) # 员工类型 — SF ``EmpJob.customString25`` picklist code (1070 正式 / # 1071 EOR / 1072 实习生 / 1073 劳务顾问 / 1074 外包 / 1075 派遣). # → Feishu ``employee_type`` enum. employee_type_code: Mapped[str | None] = mapped_column(String(16)) # 区域邮件组 / 办公室 — SF ``EmpJob.customString7`` picklist code # (2651 bj-office … 2659 sh-office … 2653 sv-office … 2865 my-office). # → Feishu ``Location`` custom attr (e.g. "SH Office"). office_code: Mapped[str | None] = mapped_column(String(16)) # 是否推送到下游系统 — SF ``EmpJob.customString24`` picklist code. # HR/SF-owned per-person downstream-provisioning flag (the same gate # Authing honoured). NOTE the field's SF picklist binding is a legacy # ``sheng`` (city) list, so its SF *labels* are meaningless (2059→ # "Valandovo", 2061→"Pehčevo"); only the CODES carry meaning. Observed # 2026-07-06: 2059=推送 (455 people, 93% already on Feishu), 2061=不推送 # (63 people, only 16% on Feishu), empty≈default-push. Stored RAW here # for review FIRST; sync-gating (exclude 不推送 from candidates) is a # SEPARATE follow-up once HR/SF confirm the code semantics. Migration 390. sf_downstream_push_code: Mapped[str | None] = mapped_column(String(16)) # GitHub username/handle — SF ``cust_Other.cust_id`` (custom MDF entity # keyed by externalCode = employee_no). Sparse (collected at onboarding); # stored RAW, junk values ("无"/"none"/empty) skipped at the sync layer. # → Feishu ``GitHub ID`` custom attr (``custom_attrs.github_id``). # Migration 369. github_id: Mapped[str | None] = mapped_column(String(128)) # Non-NULL = the CURRENT ``github_id`` was set by the employee through the # self-service change flow (github_org handle-change, migration 579) and # Helix is authoritative for it until the HRIS catches up. The HR sync # honours it: SF agreeing → write + clear the lock (HRIS is authoritative # again); SF disagreeing → keep the local value. Without it the sync's # write-when-present rule silently reverts the employee's change on the # next run, which is invisible to everyone. Migration 579. github_id_locked_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) # ── SF-verbatim labels for Google Employee_Details (migration 364) ── # Unlike the RAW picklist codes above, these hold the string SF itself # translates the code into — pushed VERBATIM to Google customSchemas # (no Helix-side value-map) so the Cloud Identity dynamic groups key on # the HRIS-authoritative value. Resolved each sync from SF picklist / # FO labels; NULL preserved on absent source (never clobbered empty). # # office_mail_group — SF ``EmailGroup`` picklist label for # ``customString7`` (e.g. 2659 → ``sh-office@pingcap.com``). office_mail_group: Mapped[str | None] = mapped_column(String(128)) # job_function — SF ``FOJobFunction.name`` for ``customString19`` # (e.g. JF0003 → ``Engineer``, JF0008 → ``IT Service``). job_function: Mapped[str | None] = mapped_column(String(128)) # job_family — SF ``cust_JobFamily.externalName`` for # ``customString21`` (e.g. JFM0002 → ``Sales-Account Executive``). job_family: Mapped[str | None] = mapped_column(String(128)) # residence_country — ISO-3166 alpha-2 of the WORK LOCATION (residence), # derived from ``work_location_code`` prefix (CHN0075 → CN). This is the # geo axis the ``g-all-china`` / ``g-all-abroad`` groups use — distinct # from ``country`` (FOCompany legal entity). 99% matches the values # Authing wrote, so taking it over is near-zero-churn. residence_country: Mapped[str | None] = mapped_column(String(8)) # Extra email addresses this person owns on the external identity # provider (typically Google Workspace aliases). Populated by # the ``email_group`` domain's Google-group mirror sync (the # legacy ``google.dept_sync`` ScheduledJob historically wrote # this column too, but was retired in PR #identity-sync-center # when its dept-projection work moved into the Identity Sync # Center). Used by cross-system sync jobs (Zoom, Slack, …) to # resolve users whose primary external-system email is one of # these aliases rather than the canonical ``work_email``. email_aliases: Mapped[list[str] | None] = mapped_column(JSON) # Dates hire_date: Mapped[date_type | None] = mapped_column(Date) termination_date: Mapped[date_type | None] = mapped_column(Date) # 生日 — sourced from SF ``PerPersonal.dateOfBirth`` (HRIS sync # writes only when current value is NULL so a person's self-edit # via ``/me/settings`` is never overwritten). Display strategy is # governed by ``identity_person_preferences`` key # ``feed.birthday_visibility`` (``month_day`` default, ``hidden`` to # suppress entirely from the feed banner / system post / lead push). # Added migration 235. birthday: Mapped[date_type | None] = mapped_column(Date) # 参加工作日期 — first day of the person's career (社会工龄 N 起点), # distinct from hire_date which is the first day at THIS company # (司龄 S 起点). Migration 163 first wired this to # ``EmpEmployment.originalStartDate``; that was wrong — SF's # ``originalStartDate`` is "first start at THIS company across # multiple stints" (司龄 anchor for rehires) and only diverges from # ``startDate`` for rehires (47/432 in prod 2026-06-02). The real # 参加工作日期 is **EmpEmployment.firstDateWorked**, populated by # PingCAP HR during onboarding for employees with prior careers # (328/500 = 66% carry it; 172 either started their career at # PingCAP or HR hasn't backfilled). The sync layer falls back # firstDateWorked → originalStartDate → user.hireDate so a row # always gets a value. NULL means HRIS didn't carry any of the # three (effectively impossible — every active employee has at # least User.hireDate). work_start_date: Mapped[date_type | None] = mapped_column(Date) # 司龄 S 起点 — SF EmpEmployment.seniorityDate. 99.8% populated in # prod survey. Diverges from ``hire_date`` only when SF tracks a # rehire / employment-segment reset (3/500 in prod 2026-06-02). # The 年假计算器's ``hire_date``-based 司龄 calculation is still # correct for 99.4% of rows; this column unlocks the rehire-edge # case and any future feature that needs the canonical 司龄 anchor. # Added migration 240. seniority_date: Mapped[date_type | None] = mapped_column(Date) # 试用期结束日 — SF EmpJob.probationPeriodEndDate. 72% populated in # prod. Used by 转正 流程, 试用期 福利年假 起算 policies, and IT 设备 # 长期分配 vs 临时分配 gating. NULL means SF doesn't carry it for # this person (typical for contractors / interns where SF skips the # probation block, or older employees imported before the field was # mandatory). Added migration 240. probation_end_date: Mapped[date_type | None] = mapped_column(Date) # FTE (full-time equivalent) — SF EmpJob.fte. 100% populated; # 99.6% are 1.0. Needed for 年假 / payroll / capacity proration # when a person is part-time (fte=0.5 etc.) — the 年假计算器 # currently hard-codes 1.0 and silently overpays half-FTE staff. # Added migration 240. fte: Mapped[Decimal | None] = mapped_column(Numeric(3, 2)) # Annual-leave balance snapshot (migration 163). Optionally synced # from SF TimeAccount; HR can override in the 年假计算器. NULL # means "no source value, HR must input". Units are days; stored as # DECIMAL(4,1) to preserve the 0.5-step rounding the policy uses. annual_leave_carry_legal_days: Mapped[Decimal | None] = mapped_column(Numeric(4, 1)) annual_leave_carry_welfare_days: Mapped[Decimal | None] = mapped_column(Numeric(4, 1)) annual_leave_used_legal_days: Mapped[Decimal | None] = mapped_column(Numeric(4, 1)) annual_leave_used_welfare_days: Mapped[Decimal | None] = mapped_column(Numeric(4, 1)) # Local password login (nullable — Google SSO is primary) hashed_password: Mapped[str | None] = mapped_column(String(255)) # Login tracking (kept here for convenience, set by auth service on login) last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Sync metadata source_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_synced_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) # HRIS tombstone (migration 222). Orthogonal to ``employment_status``: # the latter describes whether the human still works here; this one # describes whether the upstream HRIS still carries the row at all. # ``'present'`` is the normal state for every active employee and # every cleanly-offboarded leaver SF still remembers. ``'purged'`` # means a full-roster reconciliation pass found the row missing from # SF — the reconciliation handler stamps the timestamp + reason and # writes an ``hr_lifecycle_events`` row so the trail is traceable. # The row itself is NEVER hard-deleted; downstream FKs (asset # allocations, audit logs, account history) stay intact. hris_record_state: Mapped[str] = mapped_column( String(16), nullable=False, default="present", server_default="present", ) hris_purged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) hris_purged_reason: Mapped[str | None] = mapped_column(String(64)) # First-seen-absent timestamp for the daily ``hris.absent_person_check`` # future-hire cross-check (migration 421). A future hire SF's get_users # stops returning is recorded here on the FIRST run; only after the # absence persists past a grace window (≥2 consecutive daily runs) is the # row soft-deleted — so a transient SF-Onboarding flicker (P10019 陈建华) # never tombstones a real pre-hire. Cleared when SF returns them again. hris_absent_since: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Hire-day IT info push idempotency (migration 038) it_info_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) it_info_sent_channels: Mapped[list[str] | None] = mapped_column(JSON) # Day-90 onboarding-experience survey push idempotency (migration # 376) — stamped by ``hr.day90_survey_notify`` once the survey # invite (Feishu card + email) went out at ~90 days of tenure. day90_survey_sent_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) # 离职告知信推送台账 (migration 581) — stamped by # ``hr.pending_leaver_notice`` once HR's 《离职流程指引》 actually # reached the leaver (at least one channel delivered). The paired # ``offboarding_notice_last_day`` records WHICH last working day # that letter announced: a person is re-notified only when HR moves # the date in SF, never on an unchanged one. Deliberately NOT keyed # on ``OffboardingProcess`` — this letter goes out during the notice # period, weeks before any process row exists (SF only flips # ``User.status='f'`` on the last working day). offboarding_notice_sent_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) offboarding_notice_last_day: Mapped[date_type | None] = mapped_column(Date) # T+0 SSO push-logout marker (migration 090). Set when the # leaver-block runtime calls Google ``users.signOut`` / # Feishu force-logout for this person, so the audit trail shows # the exact moment the person's existing sessions were forcibly # terminated. NULL means no force-logout has happened. force_logout_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Standard timestamps created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # --- Relationships --- user_roles: Mapped[list["AuthUserRole"]] = relationship( # noqa: F821 back_populates="user", foreign_keys="AuthUserRole.user_id", lazy="selectin", ) subordinates: Mapped[list["IdentityPerson"]] = relationship( back_populates="manager", foreign_keys="IdentityPerson.manager_person_id", ) manager: Mapped["IdentityPerson | None"] = relationship( back_populates="subordinates", remote_side="IdentityPerson.id", foreign_keys="IdentityPerson.manager_person_id", ) # HRBP relationship — read-only convenience for ``person.hrbp``. We # don't expose the inverse (``hrbp_for``) because no caller needs # "all people whose HRBP is X" yet, and ``selectin`` on a high- # cardinality back-pop would over-fetch on every Person load. # ``lazy='select'`` (default) so the relationship only loads when # touched — every API request loads ``IdentityPerson`` rows on the # hot path; ``joined`` would add a self-join to every load even # though the SLA dispatcher uses ``db.get`` directly on # ``hrbp_person_id`` and never reads through this attribute. hrbp: Mapped["IdentityPerson | None"] = relationship( remote_side="IdentityPerson.id", foreign_keys="IdentityPerson.hrbp_person_id", ) org_memberships: Mapped[list["IdentityOrgMembership"]] = relationship( back_populates="person", lazy="selectin", ) external_identities: Mapped[list["IdentityExternalIdentity"]] = relationship( back_populates="person", lazy="selectin", ) # --- Convenience properties (backward compatibility) --- @property def email(self) -> str: """Backward-compatible alias for work_email.""" return self.work_email @property def status(self) -> str: """Backward-compatible alias for employment_status.""" return self.employment_status @status.setter def status(self, value: str) -> None: """Backward-compatible setter for employment_status.""" self.employment_status = value @property def department(self) -> str | None: """Backward-compatible: returns primary org membership name.""" return self.primary_department @property def primary_department(self) -> str | None: """Return the primary org_membership's org_node_name.""" for m in self.org_memberships: if m.is_primary and m.deleted_at is None: return m.org_node_name return None @property def department_chain(self) -> list[str]: """Ancestor chain from the primary org membership's ``org_path``. ``org_path`` is stored as "PingCAP / Business Operations / IT Services & Security" by the SuccessFactors sync. We strip the top-level "PingCAP" marker (it's a tenant constant — doesn't help group reports) and return everything else, leaf last. Every aggregation that bucketises by department should iterate over this chain and bump each ancestor — that's what makes a parent department show subtree totals instead of only direct reports. """ for m in self.org_memberships: if not m.is_primary or m.deleted_at is not None: continue raw = m.org_path or m.org_node_name if not raw: return [] parts = [p.strip() for p in raw.split("/") if p.strip()] # Drop the tenant root if it's present (always "PingCAP" today). if parts and parts[0].lower() in {"pingcap", "pingcap-ati"}: parts = parts[1:] # Final safety: if path was a single leaf name ensure we emit it. if not parts and m.org_node_name: parts = [m.org_node_name] return parts return [] @property def position(self) -> str | None: """Backward-compatible alias for job_title.""" return self.job_title @property def manager_id(self) -> uuid.UUID | None: """Backward-compatible alias for manager_person_id.""" return self.manager_person_id @manager_id.setter def manager_id(self, value: uuid.UUID | None) -> None: """Backward-compatible setter for manager_person_id.""" self.manager_person_id = value @property def google_sub(self) -> str | None: """Backward-compatible: find Google external identity.""" for ext in self.external_identities: if ext.provider == "google" and ext.deleted_at is None: return ext.provider_subject return None @property def employee_id(self) -> str | None: """Backward-compatible alias for employee_no.""" return self.employee_no @property def external_id(self) -> str | None: """Backward-compatible alias for the HRIS identifier.""" return self.hris_person_id @property def role_names(self) -> list[str]: # ``_role_names_override`` is set by ``get_current_user`` when the # request carries a valid ``X-View-Role`` header — i.e. when an # admin uses the header role switcher to simulate a less-privileged # role. The override is request-scoped (the IdentityPerson row is # loaded fresh per request via ``_load_user_with_roles``) and can # ONLY narrow the actual role set, never widen it: the narrowing # path validates the requested role is one the user already holds # (or is the implicit ``employee`` / ``leader`` self-downgrade). # See ``apply_view_role_narrow`` in app.core.rbac for the rules. override = self.__dict__.get("_role_names_override") if override is not None: return list(override) return normalize_role_names(ur.role.name for ur in self.user_roles) @property def actual_role_names(self) -> list[str]: """Real role set, untouched by ``X-View-Role`` narrowing. Use on the ``/auth/me`` shape and any other endpoint whose consumers (e.g. the frontend role-switcher menu) need to see which roles a user actually holds, not which role they're currently simulating. Regular RBAC checks should keep using ``role_names`` so the narrowing actually takes effect. """ return normalize_role_names(ur.role.name for ur in self.user_roles) @property def is_admin(self) -> bool: return has_role(self.role_names, GLOBAL_ADMIN_ROLE) @property def is_manager(self) -> bool: """Check if this person has any subordinates. Note: This only works if subordinates are loaded. In API endpoints, prefer using AuthService.has_subordinates() for an accurate DB query. For the asset router's inline checks, this checks the loaded subordinates relationship. """ return bool(self.subordinates) def __repr__(self) -> str: return f"" # ── Hard-gate: employment_status writes ───────────────────── # # Permitted values are `active`, `leaver_pending`, `left`, or # `pre_hire` (occasionally written by HRIS sync for hires whose # start date hasn't arrived yet — kept as a tolerated value). # Any other string is a code bug and gets rejected at assignment # time so it never reaches the DB. # # The blocker check itself (status=='left' must NOT happen while # blockers are pending) lives at TWO layers: # # 1. ``app.identity.employment_status_guard.LeaverBlockGuard`` # — explicit service every code path is required to use. # 2. ``app.identity.employment_status_listener`` — SQLAlchemy # ``before_flush`` event registered at module import that # catches direct ORM mutations bypassing the guard. # # Both layers write to ``leaver_block_audit`` on rejection. _ALLOWED_EMPLOYMENT_STATUSES = frozenset({ "active", "leaver_pending", "left", "pre_hire", }) @validates("employment_status") def _validate_employment_status(self, _key: str, value: str) -> str: if value is None: raise ValueError("employment_status cannot be NULL") normalized = str(value).strip() if normalized not in self._ALLOWED_EMPLOYMENT_STATUSES: raise ValueError( f"invalid employment_status {value!r}; " f"must be one of {sorted(self._ALLOWED_EMPLOYMENT_STATUSES)}" ) return normalized class IdentityOrgMembership(Base): __tablename__ = "identity_org_memberships" __table_args__ = ( UniqueConstraint( "person_id", "org_node_source_id", "org_node_type", name="uq_identity_org_membership", ), Index("idx_identity_org_memberships_person", "person_id"), Index("idx_identity_org_memberships_org_node", "org_node_source_id"), Index("idx_identity_org_memberships_primary", "person_id", "is_primary"), Index("idx_identity_org_memberships_parent", "parent_org_node_source_id"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=False ) source_system: Mapped[str] = mapped_column(String(32), nullable=False, default="successfactors") org_node_source_id: Mapped[str] = mapped_column(String(128), nullable=False) org_node_type: Mapped[str] = mapped_column(String(32), nullable=False, default="department") org_node_code: Mapped[str | None] = mapped_column(String(64)) org_node_name: Mapped[str] = mapped_column(String(128), nullable=False) parent_org_node_source_id: Mapped[str | None] = mapped_column(String(128)) org_path: Mapped[str | None] = mapped_column(Text) is_primary: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) membership_status: Mapped[str] = mapped_column(String(16), nullable=False, default="active") source_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_synced_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Relationships person: Mapped["IdentityPerson"] = relationship(back_populates="org_memberships") def __repr__(self) -> str: return f" {self.org_node_name}>" class IdentityOrgNode(Base): __tablename__ = "identity_org_nodes" __table_args__ = ( UniqueConstraint( "source_system", "org_node_source_id", "org_node_type", name="uq_identity_org_nodes_source", ), Index("idx_identity_org_nodes_parent", "parent_org_node_source_id"), Index("idx_identity_org_nodes_type", "org_node_type"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) source_system: Mapped[str] = mapped_column(String(32), nullable=False, default="successfactors") org_node_source_id: Mapped[str] = mapped_column(String(128), nullable=False) org_node_type: Mapped[str] = mapped_column(String(32), nullable=False, default="department") org_node_code: Mapped[str | None] = mapped_column(String(64)) org_node_name: Mapped[str] = mapped_column(String(128), nullable=False) parent_org_node_source_id: Mapped[str | None] = mapped_column(String(128)) org_path: Mapped[str | None] = mapped_column(Text) org_level: Mapped[int | None] = mapped_column() node_status: Mapped[str] = mapped_column(String(16), nullable=False, default="active") source_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_synced_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) def __repr__(self) -> str: return f"" class IdentityExternalIdentity(Base): __tablename__ = "identity_external_identities" __table_args__ = ( UniqueConstraint("provider", "provider_subject", name="uq_identity_external_provider_subject"), UniqueConstraint("person_id", "provider", name="uq_identity_external_person_provider"), Index("idx_identity_external_person", "person_id"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=False ) provider: Mapped[str] = mapped_column(String(32), nullable=False) provider_subject: Mapped[str] = mapped_column(String(255), nullable=False) provider_username: Mapped[str | None] = mapped_column(String(255)) provider_email: Mapped[str | None] = mapped_column(String(255)) is_login_identity: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) identity_status: Mapped[str] = mapped_column(String(16), nullable=False, default="active") source_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_synced_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # Relationships person: Mapped["IdentityPerson"] = relationship(back_populates="external_identities") def __repr__(self) -> str: return f"" class IdentityHrisUserAlias(Base): """A secondary HRIS ``userId`` that belongs to a person Helix already has. SuccessFactors keys a *person* by ``personIdExternal`` and an *employment* by ``userId``; Helix keys a person by ``userId`` (:attr:`IdentityPerson.hris_person_id`). Those coincide until someone holds two concurrent employments — a global assignment or a second legal entity — at which point the *organisation* hangs off whichever employment owns the team, and every HRIS reference to that manager (``EmpJob.managerId``, ``EmpJobRelationships.relUserId``) names a userId Helix does not keep a row for. One row per secondary userId, pointing at the canonical person. The UNIQUE on ``(source_system, alias_user_id)`` encodes the same invariant the primary column has: a userId identifies at most one human, ever. Resolution order is always primary column first, alias second — see ``app.identity.identity_reconcile.resolve_person_by_hris_user_id``. """ __tablename__ = "identity_hris_user_aliases" __table_args__ = ( UniqueConstraint( "source_system", "alias_user_id", name="uq_identity_hris_alias_source_user", ), Index("ix_identity_hris_alias_person", "person_id"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=False ) source_system: Mapped[str] = mapped_column( String(32), nullable=False, default="successfactors" ) alias_user_id: Mapped[str] = mapped_column(String(128), nullable=False) person_id_external: Mapped[str | None] = mapped_column(String(128)) hris_instance_id: Mapped[uuid.UUID | None] = mapped_column(Uuid) reason: Mapped[str] = mapped_column(String(32), nullable=False) note: Mapped[str | None] = mapped_column(String(255)) first_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) last_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) def __repr__(self) -> str: return f""