"""SAP SuccessFactors OData v2 client. Credentials are never read from environment variables — they live on the primary ``hris`` / ``successfactors`` row in the ``integrations`` table, managed via the admin UI at ``/admin/integrations``. Callers should construct the client asynchronously via :meth:`SuccessFactorsClient.from_config`. ``SuccessFactorsAuth`` is pluggable; swap Basic ↔ OAuth 2.0 SAML Bearer via the integration config without client code changes. """ from __future__ import annotations import asyncio import json import re from collections.abc import Callable, Iterable from datetime import UTC, date, datetime, timedelta from typing import Any import httpx from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.core.pagination import PaginationGuard from app.identity.successfactors_auth import ( SuccessFactorsAuth, build_auth_from_config, ) from app.integrations.compat import get_successfactors_config class SuccessFactorsHTTPError(RuntimeError): """SF OData endpoint returned >= 400. Carries the structured OData error code/message when the body parses as the v2 error envelope, falling back to a truncated raw body. The URL is also captured so callers / failure_notifier can pinpoint the failing entity + select fields without needing the upstream log line. """ def __init__( self, *, status_code: int, url: str, detail: str, ) -> None: self.status_code = status_code self.url = url self.detail = detail super().__init__( f"SuccessFactors HTTP {status_code} for {url} — {detail}" ) _ODATA_ERROR_BODY_CAP = 500 # SF OData v2 legacy JSON datetime — ``/Date(1719792000000)/`` (epoch ms, # optional ``+HHMM`` offset suffix which is already baked into the ms). _ODATA_MS_DATETIME_RE = re.compile(r"/Date\((?P-?\d+)(?:[+-]\d{4})?\)/") def _parse_odata_datetime(value: str | datetime | None) -> datetime | None: """Best-effort parse of SF's ``/Date(ms)/`` (or ISO) datetime strings. Local twin of ``successfactors_sync._parse_datetime`` — that module imports THIS one, so the client can't import it back. Returns ``None`` on anything unrecognized (callers treat it as "no timestamp"). """ if not value: return None if isinstance(value, datetime): return value match = _ODATA_MS_DATETIME_RE.fullmatch(value) if match: return datetime.fromtimestamp(int(match.group("ms")) / 1000, tz=UTC) try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except (ValueError, TypeError): return None # Comparisons downstream mix these with tz-aware sentinels — a naive # ISO string (no offset) would raise TypeError; pin it to UTC. return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) # Transient transport failures that warrant a bounded retry on the SF # OData GET. httpx does NOT retry on its own. The prod 2026-06-04 # ``hris.absent_person_check`` failure was a bare ``httpx.ReadTimeout`` # on the CN tenant's full ``EmpJob`` roster pull (a single slow page) — # the DE tenant succeeded the same tick, so it was transport-only, not # auth/scope/config. A single slow upstream page shouldn't fail the # whole nightly job and page every admin. Mirrors the established cloud- # driver retry set (``azure_cloud._TRANSPORT_RETRY_EXCEPTIONS``). # HTTP 4xx/5xx are intentionally NOT here: those are deterministic # (e.g. the ``positionNameEN`` 400) and surfaced via # ``SuccessFactorsHTTPError`` so retrying can't help — with the single # documented exception below (``_is_transient_auth_failure``). _TRANSPORT_RETRY_EXCEPTIONS: tuple[type[BaseException], ...] = ( httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadError, httpx.ReadTimeout, httpx.WriteError, httpx.WriteTimeout, httpx.PoolTimeout, httpx.RemoteProtocolError, ) # 3 attempts = original + 2 retries. Exponential backoff 1s, 2s (capped # at 30s, well under the worker's per-job budget). A page that's still # timing out after 3 tries is a real SF outage — let it raise so the # partial-failure notifier fires with the real ``ReadTimeout`` context # (now legible via ``format_exception_for_log`` in the handler). _MAX_TRANSPORT_ATTEMPTS = 3 _TRANSPORT_BACKOFF_CAP_SEC = 30 _MAX_PAGINATION_PAGES = 1000 _MAX_PAGINATION_ITEMS = 1_000_000 # The ONE HTTP status SF reports non-deterministically. Prod 2026-09-02 # 07:00 UTC: ``hr.pending_leaver_notice`` failed with ``401 # AuthenticationFailed: Authentication service failed to return token.`` # on ``api55.sapsf.eu/odata/v2/EmpEmployment``. That sentence is SF's own # gateway saying IAS did not hand IT a token — it is not a verdict on the # bearer we sent. Evidence it was a blip: the 02:00–06:00 ticks read the # same EU tenant fine, the CN tenant succeeded seconds earlier in the # SAME run, a live probe 4 minutes later minted a token (200, 0.98s) and # read ``EmpEmployment`` 3x on both tenants, and it is the only SF 401 in # 30 days of ``scheduled_job_runs``. Un-retried, a one-second upstream # hiccup becomes a failed run + an admin page — the false alarm that # erodes trust in the alarm that matters. A genuinely revoked credential # answers 401 too, so this only buys ~3s before the run still fails loud. _TRANSIENT_AUTH_STATUS = 401 _TRANSIENT_AUTH_MARKERS: tuple[str, ...] = ( "authenticationfailed", "authentication service failed to return token", ) def _is_transient_auth_failure(status_code: int, detail: str) -> bool: """SF's auth gateway flaked (see ``_TRANSIENT_AUTH_MARKERS``).""" if status_code != _TRANSIENT_AUTH_STATUS: return False lowered = (detail or "").lower() return any(marker in lowered for marker in _TRANSIENT_AUTH_MARKERS) def _extract_odata_error(response: httpx.Response) -> str: """Pull the OData v2 error code+message out of ``response``. OData v2 errors look like:: {"error": {"code": "...", "message": {"lang":"en","value":"..."}}} Return ``": "`` when the envelope parses; otherwise return ``raw body[:500]`` so a non-OData 400 / 500 (e.g. an HTML upstream proxy error) is still actionable. Body is hard-capped to avoid log bloat / accidental PII echoes on huge error payloads. """ try: body = response.json() except (ValueError, UnicodeDecodeError, httpx.DecodingError): # ``response.json()`` raises ``ValueError`` on malformed JSON, # ``UnicodeDecodeError`` when the declared charset doesn't match # the byte stream, and ``httpx.DecodingError`` on a broken # content-encoding (gzip truncation, etc.). All three mean # "there is a body but it isn't parseable" — fall back to the # raw text. ``response.text`` itself can also raise # ``UnicodeDecodeError`` so wrap that too. try: text = (response.text or "").strip() except UnicodeDecodeError: return "(undecodable body)" return text[:_ODATA_ERROR_BODY_CAP] or "(empty body)" err = (body or {}).get("error") if isinstance(body, dict) else None if isinstance(err, dict): code = err.get("code") or "" msg = err.get("message") if isinstance(msg, dict): value = (msg.get("value") or "").strip() elif isinstance(msg, str): value = msg.strip() else: value = "" combined = f"{code}: {value}".strip(": ").strip() if combined: return combined[:_ODATA_ERROR_BODY_CAP] # Structured-but-not-OData JSON (auth proxy, gateway, etc.) — emit a # compact repr so a top-level "message" / "fault" still surfaces. try: return json.dumps(body, ensure_ascii=False)[:_ODATA_ERROR_BODY_CAP] except (TypeError, ValueError): return repr(body)[:_ODATA_ERROR_BODY_CAP] # Shared EmpEmployment $select set — used by every list_/get_ helper that # pulls EmpEmployment rows. Centralised so adding a field (e.g. # ``firstDateWorked`` in 2026-06-02, the real 参加工作日期) only touches # one place. SF rejects unknown $select columns with HTTP 400 that tears # down the whole query, so the set MUST stay tenant-validated. # # ``firstDateWorked`` carries the person's first-ever career start (社会工龄 # N 起点) when SF has it (66% of prod rows); we previously mis-used # ``originalStartDate`` for this, which actually represents "first start at # THIS company across multiple stints" (司龄 anchor for rehires) — see # 2026-06-02 PR for the full story. # ``seniorityDate`` is SF's canonical 司龄 anchor — 99.8% populated and # diverges from ``startDate`` only on rehire / employment-segment reset. _EMP_EMPLOYMENT_SELECT_FIELDS: tuple[str, ...] = ( "personIdExternal", "userId", "startDate", "endDate", "originalStartDate", "firstDateWorked", "seniorityDate", "lastModifiedDateTime", ) class SuccessFactorsClient: def __init__( self, *, base_url: str, auth: SuccessFactorsAuth, timeout: float = 30.0, ) -> None: if not base_url: raise ValueError("SuccessFactorsClient requires a base_url") self.base_url = base_url.rstrip("/") self.auth = auth self.timeout = timeout # ── Factory ─────────────────────────────────────────────────── @classmethod async def from_config( cls, *, session: AsyncSession | None = None, timeout: float = 30.0, ) -> SuccessFactorsClient: """Build a client from the current ``successfactors`` DB config. Raises ``AppException(PLATFORM_CONFIG_MISSING)`` when the admin hasn't filled in mandatory fields yet, so the caller can surface a clear "请在配置中心填写 SuccessFactors 凭据" message. """ config = await get_successfactors_config(session=session) base_url = (config.get("base_url") or "").strip() if not base_url: raise AppException( code="INTEGRATION_NOT_CONFIGURED", message=( "SuccessFactors 连接器未配置 base_url。请到 /admin/integrations " "编辑 SuccessFactors HRIS 连接器。" ), status_code=503, ) try: auth = build_auth_from_config(config) except ValueError as e: raise AppException( code="INTEGRATION_NOT_CONFIGURED", message=f"SuccessFactors 连接器配置不完整: {e}", status_code=503, ) from e return cls(base_url=base_url, auth=auth, timeout=timeout) # ── HTTP primitives ─────────────────────────────────────────── async def _send_with_retries( self, client: httpx.AsyncClient, build_request: Callable[[], httpx.Request], ) -> httpx.Response: """Send until a < 400 response, within a bounded retry budget. Retries two things and nothing else: transient transport failures (``_TRANSPORT_RETRY_EXCEPTIONS``) and SF's own flaky auth gateway (``_is_transient_auth_failure``). Every other status raises on the first response — a deterministic 400 that retries three times is just a slower 400. Rebuild + re-apply auth per attempt: a request whose stream was partially consumed before the transport error can't be safely re-sent, and ``auth.apply`` may stamp a freshly minted / refreshed bearer. Auth is inside the try so a transport blip on the lazy token fetch retries too; a non-transport auth failure (misconfig) propagates immediately, unretried. """ last_error: BaseException | None = None for attempt in range(_MAX_TRANSPORT_ATTEMPTS): request = build_request() try: await self.auth.apply(request) response = await client.send(request) except _TRANSPORT_RETRY_EXCEPTIONS as exc: last_error = exc if attempt + 1 < _MAX_TRANSPORT_ATTEMPTS: await asyncio.sleep( min(2 ** attempt, _TRANSPORT_BACKOFF_CAP_SEC) ) continue if response.status_code < 400: return response # Surface the OData error body in the exception message — # SF's ``raise_for_status()`` default only shows status + # URL, which made the ``positionNameEN``/``positionName`` # 400 storm (2026-05-13 → 2026-05-23) un-debuggable from # logs alone. Data-endpoint bodies do NOT carry credentials # (unlike the OAuth token endpoint — see # ``successfactors_auth.py``), so it is safe to echo a # truncated snippet here. # # Strip the query string before propagating the URL into # logs / scheduled_job_runs.error: the structured ``detail`` # already names the failing property; the raw ``$filter`` # could echo OR-batched userId lists on # ``get_jobs_for_users``/``get_users`` (low-sensitivity PII # but unnecessary in an admin-readable run row). Use # ``copy_with(query=None)`` so a non-default port # (``:8443``) is preserved — staging tenants live there. error = SuccessFactorsHTTPError( status_code=response.status_code, url=str(response.request.url.copy_with(query=None)), detail=_extract_odata_error(response), ) if not _is_transient_auth_failure(error.status_code, error.detail): raise error last_error = error if attempt + 1 < _MAX_TRANSPORT_ATTEMPTS: await asyncio.sleep( min(2 ** attempt, _TRANSPORT_BACKOFF_CAP_SEC) ) # Every attempt failed transiently — re-raise the last one so # the caller's failure path + ``format_exception_for_log`` # surface the real ``ReadTimeout`` / ``SuccessFactorsHTTPError`` # class (NOT an empty string, NOT a swallowed None). assert last_error is not None raise last_error async def _get_json( self, path_or_url: str, *, params: dict[str, str] | None = None, ) -> dict[str, Any]: if path_or_url.startswith("http://") or path_or_url.startswith("https://"): # Refuse absolute pagination URLs whose host differs from base. from urllib.parse import urlparse # noqa: PLC0415 target_host = (urlparse(path_or_url).hostname or "").lower() base_host = (urlparse(self.base_url).hostname or "").lower() if not target_host or target_host != base_host: raise RuntimeError( f"SuccessFactors returned out-of-host __next URL " f"({target_host!r} != {base_host!r}); refusing " "to attach bearer cross-origin." ) target_url = path_or_url request_params: dict[str, str] | None = None else: target_url = f"{self.base_url}/{path_or_url.lstrip('/')}" request_params = params from app.integrations._url_safety import safe_async_http_transport transport = await safe_async_http_transport(target_url) async with httpx.AsyncClient( timeout=self.timeout, trust_env=False, follow_redirects=False, transport=transport, ) as client: response = await self._send_with_retries( client, lambda: client.build_request( "GET", target_url, params=request_params, ), ) return response.json() async def _post_json( self, path: str, *, json_body: Any, params: dict[str, str] | None = None, ) -> Any: """POST a JSON body to a relative OData path (e.g. ``upsert``). Mirrors ``_get_json``'s transport-retry + auth-per-attempt behaviour. Only safe for idempotent writes — the single caller today is the OData ``upsert`` endpoint, which converges on re-send by definition. """ target_url = f"{self.base_url}/{path.lstrip('/')}" from app.integrations._url_safety import safe_async_http_transport transport = await safe_async_http_transport(target_url) async with httpx.AsyncClient( timeout=self.timeout, trust_env=False, follow_redirects=False, transport=transport, ) as client: response = await self._send_with_retries( client, lambda: client.build_request( "POST", target_url, params=params, json=json_body, ), ) return response.json() async def _list_entities( self, entity_name: str, *, select_fields: Iterable[str], expand_fields: Iterable[str] | None = None, filter_clause: str | None = None, top: int | None = None, extra_params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: params = { "$format": "json", "$select": ",".join(select_fields), } if expand_fields: params["$expand"] = ",".join(expand_fields) if filter_clause: params["$filter"] = filter_clause if top is not None: params["$top"] = str(top) # SF EC effective-dating query options (``fromDate``/``toDate``/ # ``asOfDate``) are NOT ``$``-prefixed OData system params — they're # SuccessFactors navigation params that select which effective slice # of an effective-dated entity (EmpJob / EmpJobRelationships) to # return. Passed verbatim by the future-hire enrichment path so a # pre-start hire's future-dated job / work-relationship rows can be # pulled ~24h BEFORE they become the current slice. Every existing # caller omits this, so their query string is byte-identical. if extra_params: params.update(extra_params) guard = PaginationGuard( integration="successfactors", action=f"list_{entity_name}", max_pages=_MAX_PAGINATION_PAGES, max_items=_MAX_PAGINATION_ITEMS, ) def parse_page( page: object, ) -> tuple[list[dict[str, Any]], str | None]: if not isinstance(page, dict): guard.reject_malformed_response() envelope = page.get("d") if not isinstance(envelope, dict): guard.reject_malformed_response() raw_results = envelope.get("results") if ( not isinstance(raw_results, list) or any(not isinstance(item, dict) for item in raw_results) ): guard.reject_malformed_response() next_cursor = envelope.get("__next") if next_cursor is not None and not isinstance(next_cursor, str): guard.reject_malformed_response() return raw_results, next_cursor payload = await self._get_json(entity_name, params=params) page_results, next_url = parse_page(payload) guard.record_page(len(page_results)) results = list(page_results) while next_url: guard.require_safe_next(next_url) payload = await self._get_json(next_url) page_results, next_url = parse_page(payload) guard.record_page(len(page_results)) results.extend(page_results) return results # ── Read helpers ────────────────────────────────────────────── async def list_active_companies(self) -> list[dict[str, Any]]: return await self._list_entities( "FOCompany", select_fields=["externalCode", "name", "country", "status"], filter_clause="status eq 'A'", ) async def list_active_departments(self) -> list[dict[str, Any]]: return await self._list_entities( "FODepartment", select_fields=[ "externalCode", "name", "name_zh_CN", "status", "cust_orglevel", "cust_L0", "cust_L1", "cust_L2", "cust_L3", "cust_L4", "cust_L5", "parent", ], filter_clause="status eq 'A'", ) async def list_active_locations(self) -> list[dict[str, Any]]: # ``FOLocation`` on PingCAP's SF tenants exposes only # ``externalCode / name / status / timezone / locationGroup`` # — no ``city`` / ``country`` / address columns. The ``name`` # field carries the city-level label (e.g. ``Delhi``, # ``Mumbai``, ``Beijing``), so the HR service maps it onto # ``identity_persons.city``. Country comes from # ``FOCompany.country`` (ISO-3) via ``EmpJob.company``. return await self._list_entities( "FOLocation", select_fields=["externalCode", "name", "status"], filter_clause="status eq 'A'", ) async def list_github_ids(self) -> list[dict[str, Any]]: """List the ``cust_Other`` custom MDF entity rows. ``externalCode`` is the employee_no; ``cust_id`` carries the employee's GitHub username/handle (collected on the onboarding 「其他声明」form — sparse, populated only for those who filled it, verified live 2026-06-30). Junk values ("无"/"none"/empty) are filtered downstream in the HR sync before they reach the person row. No status filter — the MDF query returns current records. """ return await self._list_entities( "cust_Other", select_fields=["externalCode", "cust_id"], ) async def list_active_positions(self) -> list[dict[str, Any]]: return await self._list_entities( "Position", select_fields=[ "code", "externalName_defaultValue", "effectiveStatus", "targetFTE", "department", "company", "location", "positionControlled", "multipleIncumbentsAllowed", "parentPosition/code", "parentPosition/externalName_defaultValue", ], expand_fields=["parentPosition"], filter_clause="effectiveStatus eq 'A'", ) async def list_job_functions(self) -> list[dict[str, Any]]: """List ``FOJobFunction`` rows (Job Function foundation object). ``EmpJob.customString19`` carries the ``JF000x`` externalCode; this resolves it to the authoritative SF label (e.g. ``JF0003`` → ``Engineer``, ``JF0008`` → ``IT Service``). The label is used verbatim — Helix maintains no code→label table. """ return await self._list_entities( "FOJobFunction", select_fields=["externalCode", "name", "name_zh_CN", "status"], filter_clause="status eq 'A'", ) async def list_job_families(self) -> list[dict[str, Any]]: """List ``cust_JobFamily`` rows (Job Family MDF object). ``EmpJob.customString21`` carries the ``JFM00xx`` externalCode; this resolves it to the SF ``externalName`` (e.g. ``JFM0002`` → ``Sales-Account Executive``). MDF object, so ``externalName`` (not ``name``) is the label column. """ return await self._list_entities( "cust_JobFamily", select_fields=["externalCode", "externalName"], ) async def list_email_group_labels(self) -> dict[str, str]: """Resolve the ``EmailGroup`` picklist: option id → label. ``EmpJob.customString7`` is the office picklist option **id** (e.g. ``2659``); the SF ``EmailGroup`` picklist translates it to the office mail-group address verbatim (``2659`` → ``sh-office@pingcap.com``). Helix stores no hardcoded office map — this is the single source of truth, refreshed every sync. The bulk ``$expand=picklistLabels`` on PicklistOption drops the labels on this tenant, so we read the ``PicklistLabel`` entity per option (≈12 options — cheap, once per sync). Prefers the ``en_US`` locale, falling back to any available locale. """ options = await self._list_entities( "PicklistOption", select_fields=["id"], filter_clause="picklist/picklistId eq 'EmailGroup'", ) out: dict[str, str] = {} for opt in options: oid = str(opt.get("id") or "").strip() if not oid: continue labels = await self._list_entities( "PicklistLabel", select_fields=["optionId", "locale", "label"], filter_clause=f"optionId eq {oid}", ) by_locale = {row.get("locale"): row.get("label") for row in labels} label = by_locale.get("en_US") or next(iter(by_locale.values()), None) if label: out[oid] = label return out async def list_current_jobs( self, limit: int | None = None, ) -> list[dict[str, Any]]: """List every active EmpJob row. ``limit=None`` (default) fetches the full active-jobs roster, following ``__next`` links to exhaustion. Pass an explicit positive ``limit`` only when truncation is intentional (smoke tests, debug fetches). Pre-launch P1 fix (round-19 audit, 2026-05-10): the prior default of ``limit=10`` was a footgun — OData's ``$top=10`` instructs the server to return at most 10 rows total, so any caller that forgot to pass an explicit ``None`` would silently sync only the first 10 employees. Today's only callers (``identity.successfactors_sync.sync_users`` and the scheduled sync handler) already pass an explicit value, but the default is the kind of gun the next refactor leaves loaded. """ # Field set MUST mirror ``get_jobs_for_users`` — that is the only # field list this client has empirically observed PingCAP's SF # tenant (api*.sapsf.cn) accept on ``EmpJob``. Adding fields that # aren't provisioned on the tenant makes OData return HTTP 400 # with ``Property '' is invalid``; the entire query fails # before any row is returned, so absent_person_check / full # roster sync break silently in prod. # # Specifically DO NOT add ``positionNameEN`` / ``positionName`` # back — those aren't standard EmpJob properties and the # PingCAP tenant rejects them (broke ``hris.absent_person_check`` # daily for 10 days, 2026-05-13 → 2026-05-23). ``jobTitle`` is # standard per the SAP OData spec but was never validated on # this tenant either; downstream consumers (sync_users + # _apply_event) already fall back to ``Position.externalName_ # defaultValue`` resolved via the master-data cache, so dropping # it here is safe. return await self._list_entities( "EmpJob", select_fields=[ "userId", "company", "department", "location", "managerId", "position", "emplStatus", "effectiveLatestChange", "startDate", "endDate", "costCenter", # Added 2026-06-02 — fte + probationPeriodEndDate land # on ``identity_persons`` (migration 240) for 年假 / 转正 # downstream usage. Both confirmed accepted by PingCAP's # SF tenant in the 2026-06-02 SF dump survey. "fte", "probationPeriodEndDate", # Added migration 344 — projected to Feishu via the # identity-sync ``feishu_contact`` target: # customString7 → 区域邮件组 / 办公室 (Feishu Location) # customString25 → 员工类型 (Feishu employee_type) # Both confirmed accepted by PingCAP's SF tenant (live # $select probe 2026-06-26). Stored RAW on identity_persons. "customString7", "customString25", # customString16 → 实际工作国家 (actual work country, ISO-3 # like CHN/JPN/USA — HR-confirmed 2026-07-02). Feeds # ``identity_persons.country`` (precedence over the company # entity country) so Feishu Region + mobile dialing reflect # where the person actually works, not the payroll entity. "customString16", # customString24 → 是否推送到下游系统 (HR/SF-owned per-person # downstream-provisioning flag; codes 2059=推送 / 2061=不推送, # its picklist labels are legacy-garbage). Stored RAW for # review; gating is a separate follow-up. Migration 390. "customString24", # customString15 → 【是否主岗】 (2059=Yes / 2061=No), # located 2026-09-04 once HR had ticked "No" on a real 兼岗. # ⚠️ SAME Yes/No picklist as customString24 above, so the ids # are shared — `2061` means nothing until you know WHICH # field it came from. Read only as an explicit No (the # picklist is nullable); see app/identity/sf_assignment.py. "customString15", # Job Function / Job Family FO externalCodes (verified # 2026-06-29). customString19 → FOJobFunction (JF000x → # 'Engineer' …), customString21 → cust_JobFamily (JFM00xx # → 'Sales-Account Executive' …). Resolved to SF labels in # the sync and pushed verbatim to Google Employee_Details. "customString19", "customString21", # International assignment / 外派 (field IDs confirmed via # live $metadata labels 2026-07-07). Supplementary context # for HR reports / person detail (does NOT repoint directory # placement): # customString44 → 外派前国家 (home country, ISO-3) # customString38 → 外派后国家 (host country, ISO-3) # customString39 → 外派状态 (status, RAW picklist code) # customDate2 → 外派开始日期 # customDate4 → 外派结束日期 "customString44", "customString38", "customString39", "customDate2", "customDate4", ], filter_clause="effectiveLatestChange eq true", top=limit, ) async def list_job_relationships( self, relationship_types: list[str], user_ids: list[str] | None = None, *, effective_as_of: date | None = None, ) -> list[dict[str, Any]]: """List ``EmpJobRelationships`` rows for the given relationship types. Backs the HRBP / Budget-Owner (调薪负责人) dept-role sync. Each row is ``{userId, relationshipType, relUserId}`` — the per-person work relationship (工作关系). ``relationship_types`` are the SF tenant's relationshipType option codes (PingCAP: 2448=HRBP, 2447=调薪负责人; verified live 2026-06-16). Filtered server-side to just those types so the pull stays small (≈2 rows/employee). ``user_ids`` (optional) additionally scopes the pull to those SF userIds — used by the live onboard-webhook path so a SINGLE hire resolves its HRBP / HR Ops with a ~3-row query instead of dragging the whole tenant's relationship feed. When omitted the query is tenant-wide (the full-roster confirm/bootstrap path). Returns ``[]`` (NOT raise) on any failure — this is an optional enrichment; a missing/permission-denied EmpJobRelationships entity must never break the core EmpJob roster sync. The caller logs the degradation. """ if not relationship_types: return [] # OData v2 OR-of-equals on relationshipType (a String option code). type_clause = " or ".join( f"relationshipType eq '{t}'" for t in relationship_types ) clause = f"({type_clause})" if user_ids: uid_clause = " or ".join(f"userId eq '{u}'" for u in user_ids) clause = f"{clause} and ({uid_clause})" return await self._list_entities( "EmpJobRelationships", select_fields=["userId", "relationshipType", "relUserId"], filter_clause=clause, top=None, extra_params=_effective_asof_params(effective_as_of), ) async def list_hire_employments( self, start_date: date, end_date: date, ) -> list[dict[str, Any]]: return await self._list_entities( "EmpEmployment", select_fields=_EMP_EMPLOYMENT_SELECT_FIELDS, filter_clause=( f"startDate ge { _odata_datetime_literal(start_date) } " f"and startDate lt { _odata_datetime_literal(end_date + timedelta(days=1)) }" ), ) async def list_termination_employments( self, start_date: date, end_date: date, ) -> list[dict[str, Any]]: return await self._list_entities( "EmpEmployment", select_fields=_EMP_EMPLOYMENT_SELECT_FIELDS, filter_clause=( "endDate ne null " f"and endDate ge { _odata_datetime_literal(start_date) } " f"and endDate lt { _odata_datetime_literal(end_date + timedelta(days=1)) }" ), ) async def get_employments_for_users( self, user_ids: list[str], ) -> list[dict[str, Any]]: """Fetch ``EmpEmployment`` rows for a specific list of userIds. Used by the bootstrap full sync to pin down ``endDate`` (and therefore ``termination_date``) for SF users whose ``status`` is ``'f'`` (forfeited). The date-window helpers (``list_termination_employments`` / ``list_hire_employments``) require knowing the date upfront — but for an arbitrary historical backfill we know only "this user is terminated, find when". Chunked by 50 like every other ``... eq ... or`` query so a 500-user forfeited list doesn't blow SF's URL limit. """ if not user_ids: return [] results: list[dict[str, Any]] = [] for chunk in _chunked(user_ids, size=50): filter_clause = " or ".join( f"userId eq '{_odata_str(user_id)}'" for user_id in chunk ) chunk_results = await self._list_entities( "EmpEmployment", select_fields=_EMP_EMPLOYMENT_SELECT_FIELDS, filter_clause=filter_clause, ) results.extend(chunk_results) return results async def get_termination_reasons_for_users( self, user_ids: list[str], ) -> dict[str, str]: """userId → termination-reason code from ``EmpEmploymentTermination``. PingCAP's tenant stores the 离职原因 on the custom field ``EmpEmploymentTermination.customString85`` — a picklist (``EcEvenReason2``) whose RAW value is the picklist **option id** (e.g. ``2567``), NOT the externalCode (``D04``) the HR team quotes. ``$expand=customString85Nav`` resolves the option to its ``externalCode`` in the same round-trip (verified on both api30.sapsf.cn and api55.sapsf.eu, 2026-07-30 — option ids AND externalCodes are identical across the two tenants). Returns the externalCode per userId when the expand works, and degrades to the raw option id when the tenant rejects the expand — callers' exclusion config should therefore list externalCodes (``D03``) and MAY additionally pin raw option ids (``2566``) as belt-and-suspenders. A userId can carry several termination rows (rehire history). The row with the latest ``lastModifiedDateTime`` wins; when the tenant rejects that field the last row returned wins (single-row per user is the overwhelmingly common case). """ if not user_ids: return {} out: dict[str, str] = {} best_lmd: dict[str, datetime] = {} # Progressive degrade per chunk: (select with lastModifiedDateTime # + expand) → (select without lmd + expand) → (raw option id, no # expand). SF rejects unknown $select/$expand members with HTTP # 400 for the WHOLE query, so each fallback strictly shrinks the # field set. attempts: list[tuple[list[str], list[str] | None]] = [ ( ["userId", "customString85", "lastModifiedDateTime", "customString85Nav/externalCode"], ["customString85Nav"], ), ( ["userId", "customString85", "customString85Nav/externalCode"], ["customString85Nav"], ), (["userId", "customString85"], None), ] for chunk in _chunked(sorted(set(user_ids)), size=50): filter_clause = " or ".join( f"userId eq '{_odata_str(user_id)}'" for user_id in chunk ) rows: list[dict[str, Any]] | None = None last_exc: Exception | None = None for select_fields, expand_fields in attempts: try: rows = await self._list_entities( "EmpEmploymentTermination", select_fields=select_fields, expand_fields=expand_fields, filter_clause=filter_clause, ) break except SuccessFactorsHTTPError as exc: if exc.status_code != 400: # Only a 400 means "this tenant rejects this # $select/$expand shape". A 401 / 403 / 5xx says # nothing about the shape — degrading on it walks # the two narrower shapes against the same broken # endpoint and then reports the NARROWEST one's # failure, so the run row blames a $select that # was never the problem (prod 2026-09-02: SF's # auth gateway 401'd mid-job). raise last_exc = exc continue if rows is None: # All shapes rejected — surface the failure; the caller # decides whether reason lookup is best-effort. assert last_exc is not None raise last_exc for row in rows: uid = row.get("userId") if not uid: continue nav = row.get("customString85Nav") or {} code = str( nav.get("externalCode") or row.get("customString85") or "" ).strip() if not code: continue lmd = _parse_odata_datetime( row.get("lastModifiedDateTime") ) or datetime.min.replace(tzinfo=UTC) if uid not in out or lmd >= best_lmd.get( uid, datetime.min.replace(tzinfo=UTC) ): out[uid] = code best_lmd[uid] = lmd return out async def get_jobs_for_users( self, user_ids: list[str], *, effective_as_of: date | None = None, ) -> list[dict[str, Any]]: """Fetch ``EmpJob`` rows for the given userIds. ``effective_as_of`` (future-hire enrichment only): pass a date to request the effective slice AS OF that date instead of SF's default as-of-today. A pre-start hire's job row (department / managerId / company) is future-dated, so the default slice returns nothing until the hire date arrives; passing the hire date lets the 24h-lead build pull it early. Every other caller omits it and keeps the current as-of-today behaviour (byte-identical query string). """ if not user_ids: return [] results: list[dict[str, Any]] = [] asof_params = _effective_asof_params(effective_as_of) for chunk in _chunked(user_ids, size=50): filter_clause = " or ".join(f"userId eq '{_odata_str(user_id)}'" for user_id in chunk) chunk_results = await self._list_entities( "EmpJob", # Keep in sync with ``list_current_jobs``: the downstream # confirm pipeline needs company/department/location/manager # to populate IdentityPerson.job_title + org memberships + # manager_person_id. Without these, the new batch sync # leaves every synced person with NULL job_title (seen # 2026-04-21 — 0/501 filled). select_fields=[ "userId", "startDate", "endDate", "event", "eventReason", "emplStatus", "lastModifiedDateTime", "seqNumber", "company", "department", "location", "managerId", "position", # ``costCenter`` is the SF EmpJob attribute holding # the cost-center code. Pulled to populate # ``IdentityPerson.cost_center`` so the Identity # Sync Center can project it into Navan / SCIM # enterprise.costCenter. "costCenter", # Added 2026-06-02 — fte + probationPeriodEndDate # land on ``identity_persons`` (migration 240) for # 年假 / 转正 downstream usage. Mirrors the # ``list_current_jobs`` field set. "fte", "probationPeriodEndDate", # Migration 344 — keep in sync with # ``list_current_jobs``. customString7 → 区域邮件组/ # 办公室, customString25 → 员工类型; both projected to # Feishu by the feishu_contact identity-sync target. "customString7", "customString25", # customString16 → 实际工作国家 (actual work country) — # keep in sync with ``list_current_jobs``. Feeds # identity_persons.country (feishu Region + mobile). "customString16", # customString24 → 是否推送到下游系统 (downstream flag) — # keep in sync with ``list_current_jobs``. Stored RAW. "customString24", # customString15 → 【是否主岗】 — keep in sync with # ``list_current_jobs``. Explicit-No only; see # app/identity/sf_assignment.py. "customString15", # Job Function / Job Family FO codes — keep in sync # with ``list_current_jobs`` (Google Employee_Details). "customString19", "customString21", # International assignment / 外派 (migration 392) — keep # in sync with ``list_current_jobs`` so a person synced via # either path lands the same fields. IDs confirmed via live # $metadata labels: customString44=外派前国家, # customString38=外派后国家, customString39=外派状态, # customDate2/4=外派开始/结束日期. "customString44", "customString38", "customString39", "customDate2", "customDate4", ], filter_clause=filter_clause, extra_params=asof_params, ) results.extend(chunk_results) return results async def get_users(self, user_ids: list[str]) -> list[dict[str, Any]]: if not user_ids: return [] results: list[dict[str, Any]] = [] for chunk in _chunked(user_ids, size=50): id_clause = " or ".join(f"userId eq '{_odata_str(user_id)}'" for user_id in chunk) # Constrain status explicitly — PingCAP's tenant applies an # implicit ``status='t'`` filter when ``$filter`` omits status # (see ``_USER_STATUS_ANY_FILTER``), so a bare userId-only # filter silently drops forfeited (``'f'``) and future-hire # (``active_external_suite``) users. Same contract trap fixed # for ``list_all_users`` / ``list_users_modified_since`` on # 2026-07-12; ``get_users`` (preview_import + the webhook path) # was missed and returned ``None`` for a future hire, staging # them with a synthetic ``@successfactors.local`` identity. filter_clause = f"({self._USER_STATUS_ANY_FILTER}) and ({id_clause})" chunk_results = await self._list_entities( "User", select_fields=[ "userId", "displayName", "email", "empId", "status", "hireDate", "lastModifiedDateTime", # See ``_USER_SELECT_FIELDS`` comment (migration 240). "firstName", "lastName", # LEGACY fallback for the HR-assigned corporate 邮箱名 # (email local-part, e.g. "john.chen"). The AUTHORITATIVE # source is now ``PerPersonal.customString4`` (see # ``get_personal_email_names`` — matches SF's own outbound # ``emailName <<-- customString4`` Integration Center # mapping). ``custom07`` (a DIFFERENT entity + field) only # coincidentally held the value on PingCAP's tenant # (verified 2026-07-20 P10019="john.chen"); kept as a # defensive secondary so a person with a blank customString4 # still resolves. Both feed reserve_work_email_for above the # given.family pinyin fallback. "custom07", ], filter_clause=filter_clause, ) results.extend(chunk_results) return results # Column set shared by ``list_active_users`` / ``list_all_users`` / # ``list_users_modified_since``. SF's User entity rejects unknown # ``$select`` columns with HTTP 400 (and tears down the whole query), # so the set MUST stay minimal and tenant-validated. Adding here # ripples to every caller — pin the diff to one place. _USER_SELECT_FIELDS = ( "userId", "displayName", "email", "empId", "status", "hireDate", "lastModifiedDateTime", # Added 2026-04-22 — preferred_language pulled into # identity_persons (migration 048). ``defaultLocale`` is a # standard User field (values like ``en_US`` / ``zh_CN``). # NOTE: ``personalMobile`` / ``personalEmail`` are NOT fields on # SF User — personal contact lives on PerPhone / PerEmail # entities (see ``list_personal_phones`` / ``list_personal_emails``). "defaultLocale", # Added 2026-06-02 (migration 240) — authoritative given/family # split so the Google account provisioner can build the email # local-part as ``given.family`` without guessing the family # boundary from pinyin tokens. Standard SF User columns; both # land on ``identity_persons.first_name`` / ``last_name``. "firstName", "lastName", # LEGACY fallback for the HR-assigned corporate 邮箱名 (email # local-part, e.g. "john.chen"). AUTHORITATIVE source is now # ``PerPersonal.customString4`` (see ``get_personal_email_names``); # ``custom07`` is a different entity+field kept only as a defensive # secondary. Staged as ``email_name`` when customString4 is blank so # reserve_work_email_for still gets a name above the pinyin fallback. "custom07", ) # Every ``User.status`` value the PingCAP tenants are known to emit. # SF applies an implicit ``status='t'`` filter whenever ``$filter`` # does not constrain ``status`` (verified 2026-06-01), so any roster # or incremental query MUST spell the full set out. Third value # ``active_external_suite`` discovered 2026-07-12: future hires # entered through SF Onboarding carry it until day 1 — they have a # real future-dated EmpJob/EmpEmployment but were invisible to every # two-status query, so an entire EU cohort never reached Helix and # nobody got accounts on their start date. Extend HERE when SF # grows a new status; every roster consumer picks it up. _USER_STATUS_ANY_FILTER = ( "status eq 't' or status eq 'f' or status eq 'active_external_suite'" ) async def list_active_users(self) -> list[dict[str, Any]]: """Enumerate every **active** SF user (``status eq 't'``). Use this where the caller specifically needs the current in-roster set — most importantly ``run_hris_absent_person_check``, which diffs SF's live active set against Helix's active set to find people SF lost. Do NOT use it for the bootstrap full sync — that path needs leavers too (see :meth:`list_all_users`). """ return await self._list_entities( "User", select_fields=list(self._USER_SELECT_FIELDS), filter_clause="status eq 't'", ) async def list_all_users(self) -> list[dict[str, Any]]: """Enumerate **every** SF user regardless of status. Returns both ``status='t'`` (active) and ``status='f'`` (forfeited / terminated) users in one shot. Used by the full-employee bootstrap sync so a freshly-deployed Helix instance imports historical leavers too, not just the current roster. Why bootstrap can't just use ``list_active_users``: incremental sync only ever fetches rows whose ``lastModifiedDateTime`` is after the previous run cursor. Anyone SF flipped to ``'f'`` before the bootstrap timestamp falls in a permanent blind spot — their last modification is in the past, and ``status eq 't'`` filters them out of the catch-up roster. The result is silently anchor-less downstream data (asset / DNS / SaaS rows whose owner string still names them but whose ``owner_person_id`` is NULL because Helix never knew the person existed). Pre-launch P0 fix (2026-05-30): PingCAP prod had 48 SF leavers in this blind spot — wangdi@pingcap.com (P0175, terminated 2026-03-20) was the user-reported case. See the doc at ``docs/architecture/PLATFORM.md`` §HR/SF-sync for the post-mortem. Follow-up fix (2026-06-01): the original docstring claimed an unfiltered ``$filter``-less query returns active + forfeited in one page. That assumption is FALSE on PingCAP's tenant — the first prod run of ``hris.full_sync`` (PR #635) staged 434 events, all of which were active users; ZERO forfeited rows came back. Empirically SF applies an implicit "active only" filter when none is given. Calling ``$filter=status eq 'f'`` explicitly returns the 48 forfeited rows (verified by the same-day backfill-script dry-run). The fix: pass an explicit ``status eq 't' or status eq 'f'`` so both buckets are unambiguously included regardless of SF tenant defaults. """ return await self._list_entities( "User", select_fields=list(self._USER_SELECT_FIELDS), # Explicit status OR — required because PingCAP's SF tenant # returns only ``status='t'`` when ``$filter`` is omitted # (empirically verified 2026-06-01). Shared constant so new # tenant statuses (e.g. 2026-07-12 ``active_external_suite``) # extend every roster consumer at once. filter_clause=self._USER_STATUS_ANY_FILTER, ) # ── Incremental sync helpers ───────────────────────────────── async def list_users_modified_since( self, since: datetime, ) -> list[dict[str, Any]]: """Users whose ``lastModifiedDateTime`` is ≥ ``since``. Covers both new hires and leavers: SF increments ``lastModifiedDateTime`` whenever any user attribute changes (including ``status`` flipping from ``'t'`` to ``'f'`` on termination). Used by the scheduled incremental sync so the daily tick picks up *changes since last run* rather than replaying the whole roster. Leaver blind-spot fix (2026-07-03, P0051 韦万 post-mortem): PingCAP's SF tenant applies an implicit ``status='t'`` filter whenever the ``$filter`` clause does NOT constrain ``status`` — the exact trap ``list_all_users`` was explicitly fixed for on 2026-06-01 (PR #635). A ``lastModifiedDateTime``-only filter hits the same default, so a user whose ``status`` just flipped ``t→f`` is silently dropped from the incremental feed even though their modification is squarely inside the window. The leaver then never gets staged as an offboard and stays ``active`` in Helix until the weekly full sync. Constrain ``status`` explicitly (both buckets) so forfeited users are unambiguously included regardless of tenant defaults — mirrors ``list_all_users``. If SF ever adds a third status code, extend this clause rather than dropping back to a status-less filter. """ return await self._list_entities( "User", select_fields=list(self._USER_SELECT_FIELDS), filter_clause=( f"({self._USER_STATUS_ANY_FILTER}) and " f"lastModifiedDateTime ge { _odata_datetime_literal(since) }" ), ) # ── Personal contact helpers (PerPhone / PerEmail) ─────────── # # Personal mobile + personal email live on SF's PerPhone / PerEmail # entities, keyed by ``personIdExternal`` (NOT ``userId``). The HR # sync pipeline maps personIdExternal → userId via the EmpEmployment # rows it already fetches, so these two helpers don't need to know # about User at all. # # Picklist codes are tenant-configured. The values below are what # PingCAP's CN + EU SF instances use as of 2026-04 (confirmed via # metadata probe): # PerPhone.phoneType = '1043' → Mobile / Cell # PerEmail.emailType = '1047' → Personal email (private domains) # If a new tenant is onboarded with different codes, override via # the integration config rather than hardcoding here. async def list_personal_phones( self, *, phone_type: str = "1043", ) -> list[dict[str, Any]]: """List PerPhone rows for the given phoneType code. Default ``1043`` is "Mobile/Cell" on PingCAP's SF tenants. """ return await self._list_entities( "PerPhone", select_fields=[ "personIdExternal", "phoneType", "phoneNumber", # ``countryCode`` is a ``CountryCode`` picklist OPTION id # (e.g. 1734), NOT an ISO/dialing code. Resolve to ISO-2 via # :meth:`list_country_code_picklist`. Carries the phone's OWN # country so a host-country phone on an international assignee # (home CN, phone +65 SG) formats with the right dialing code # instead of the home-country default. "countryCode", "isPrimary", ], filter_clause=f"phoneType eq '{phone_type}'", ) async def list_user_emails( self, user_ids: list[str], ) -> dict[str, str]: """Resolve ``userId → User.email`` for the given userIds (batched). Used by the cross-instance personal-phone merge (hr sync) to guard same-person identity: SF ``userId`` is NOT unique across the CN (api30) and EU (api55) tenants, so a phone pulled from one instance for a person owned by the other is only trusted when that instance's ``User.email`` matches the Helix person's work_email. Returns only non-empty emails; keyed by ``userId``. Chunked by 40 like every other ``... eq ... or`` query so a large batch doesn't blow SF's URL-length limit. """ if not user_ids: return {} out: dict[str, str] = {} for chunk in _chunked(sorted(set(user_ids)), size=40): clause = " or ".join( f"userId eq '{_odata_str(u)}'" for u in chunk ) rows = await self._list_entities( "User", select_fields=["userId", "email"], filter_clause=clause, ) for r in rows: uid = r.get("userId") em = r.get("email") if uid and isinstance(em, str) and em.strip(): out[uid] = em.strip() return out async def list_country_code_picklist(self) -> dict[str, str]: """``CountryCode`` picklist: option id → ISO-3166 alpha-2. SF stores ``PerPhone.countryCode`` as a picklist OPTION id; the matching :class:`PickListValueV2` row on ``PickListV2_id='CountryCode'`` carries the ISO-2 in ``externalCode`` (verified live 2026-07-07: optionId ``1734`` → ``externalCode='SG'``, label ``+65 (SG)``). Returns ``{str(optionId): ISO2}`` so callers translate a phone's raw country option into the ISO-2 the phone formatter keys on. Reads only. A picklist fetch failure is the caller's to tolerate (degrade to home-country derivation) — we don't swallow here. """ rows = await self._list_entities( "PickListValueV2", select_fields=["optionId", "externalCode"], filter_clause="PickListV2_id eq 'CountryCode'", ) out: dict[str, str] = {} for r in rows: opt = r.get("optionId") ext = r.get("externalCode") if opt is not None and isinstance(ext, str) and ext.strip(): out[str(opt)] = ext.strip().upper() return out async def list_personal_birthdays( self, ) -> list[dict[str, Any]]: """List PerPersonal master rows (birthday + gender). PingCAP's SF tenant stores employee date-of-birth on the ``PerPersonal`` extension entity in a custom field — NOT on the canonical ``User.dateOfBirth`` field (which is null for every user in this tenant; verified 2026-06-01). The custom field is ``customDate1``. ``gender`` (``M`` / ``F``) is a standard PerPersonal column, added migration 344 — projected to Feishu's ``gender`` enum by the feishu_contact identity-sync target. Confirmed accepted by PingCAP's tenant in the same $select (live probe 2026-06-26). Returns rows of ``{personIdExternal, customDate1, gender}`` so the caller can map ``hris_person_id → birthday`` / ``gender`` from a single master-data pull (the method name is kept for backward compatibility — it has always been the PerPersonal master fetch). """ return await self._list_entities( "PerPersonal", select_fields=[ "personIdExternal", "customDate1", "gender", ], ) async def get_personal_email_names( self, person_id_externals: list[str], ) -> dict[str, str]: """Resolve ``personIdExternal → HR-assigned 邮箱名`` from PerPersonal. The desired corporate mailbox local-part (例：``san.zhang``) is authoritatively stored by HR on the SF ``PerPersonal`` extension entity, field ``customString4`` — confirmed by SF's own Integration Center outbound mapping (``Add New Employee`` ships ``emailName <<-- customString4 from PerPersonal``). The push (webhook) path already receives that value as ``emailName``; this method is the PULL-path (manual/full/incremental sync) counterpart so both routes source the local-part from the SAME SF field. Superseded the earlier ``User.custom07`` read (a different entity + field that only coincidentally held the value on PingCAP's tenant) — see ``get_users`` `$select` note. Keyed by ``personIdExternal`` (NOT ``userId``); callers map via the ``EmpEmployment`` rows they already fetch. Chunked by 50 like every other ``... eq ... or`` query so a 500-person batch doesn't blow SF's URL length limit. Returns only non-empty, stripped values — an absent / blank ``customString4`` is simply omitted so the caller falls back to the existing derivation. """ if not person_id_externals: return {} out: dict[str, str] = {} for chunk in _chunked(sorted(set(person_id_externals)), size=50): id_clause = " or ".join( f"personIdExternal eq '{_odata_str(pid)}'" for pid in chunk ) rows = await self._list_entities( "PerPersonal", select_fields=["personIdExternal", "customString4"], filter_clause=id_clause, ) for row in rows: pid = row.get("personIdExternal") name = row.get("customString4") if isinstance(pid, str) and isinstance(name, str) and name.strip(): out[pid] = name.strip() return out async def list_personal_emails( self, *, email_type: str = "1047", ) -> list[dict[str, Any]]: """List PerEmail rows for the given emailType code. Default ``1047`` is "Personal" (non-corporate domain) on PingCAP's SF tenants — business email is ``1046`` and is already carried on User.email. """ return await self._list_entities( "PerEmail", select_fields=[ "personIdExternal", "emailType", "emailAddress", "isPrimary", ], filter_clause=f"emailType eq '{email_type}'", ) async def get_person_id_external(self, user_id: str) -> str | None: """Resolve a SF ``userId`` to its ``personIdExternal``. PerEmail (and the other Per* person entities) are keyed by ``personIdExternal``, NOT ``userId`` — Helix only persists the latter (``identity_persons.hris_person_id``), so a write to PerEmail must hop through EmpEmployment first. Returns ``None`` when SF has no employment row for the userId; callers must fail closed rather than guess (writing someone else's email onto the wrong person is worse than not writing back at all). """ if not user_id: return None escaped = user_id.replace("'", "''") rows = await self._list_entities( "EmpEmployment", select_fields=["personIdExternal", "userId"], filter_clause=f"userId eq '{escaped}'", top=1, ) for row in rows: pid = (row or {}).get("personIdExternal") if pid: return str(pid) return None async def upsert_per_email( self, *, person_id_external: str, email_address: str, email_type: str, is_primary: bool = True, ) -> dict[str, Any]: """Upsert one PerEmail row via the OData ``upsert`` endpoint. Payload shape matches the sample HR's identity team provided (2026-06-11):: POST {base_url}/upsert?$format=json [{"__metadata": {"uri": "PerEmail"}, "emailType": "1046", "personIdExternal": "...", "emailAddress": "...", "isPrimary": true}] SF returns HTTP 200 even for record-level failures — the real verdict lives in the per-record envelope (``status`` / ``httpCode`` / ``message``). This method returns that first record dict; the caller decides success from it. HTTP-level (>= 400) failures raise ``SuccessFactorsHTTPError`` as usual. """ records = await self.upsert_per_emails([ { "emailType": email_type, "personIdExternal": person_id_external, "emailAddress": email_address, "isPrimary": is_primary, } ]) if records and isinstance(records[0], dict): return records[0] return {"status": "UNKNOWN", "raw": records} async def upsert_per_emails( self, records: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Upsert MANY ``PerEmail`` rows in ONE atomic OData ``upsert``. SF allows only ONE primary email per person, and rejects a second ``isPrimary=true`` while another row still holds it ("只能将一个 电子邮箱地址设置为主要电子邮箱地址"). To make the corporate email primary we must therefore promote it AND demote the prior primary in the SAME batch — SF validates the batch as a whole, so the net state (exactly one primary) is accepted. ``records`` are logical dicts (``emailType`` / ``personIdExternal`` / ``emailAddress`` / ``isPrimary``); the ``__metadata`` wrapper is added here. Returns the per-record envelopes in order; the caller decides success. """ return await self.upsert_records("PerEmail", records) async def upsert_records( self, entity: str, records: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Upsert rows of ANY OData entity in ONE atomic ``upsert`` call. The entity-agnostic core :meth:`upsert_per_emails` was built on and now delegates to. Extracted when the GitHub-ID write-back needed the same call shape against the ``cust_Other`` MDF entity: identical transport, identical envelope quirks, only the ``__metadata.uri`` and the field names differ — copying the body would have meant two places to fix the next time SF changes its envelope. ``records`` are logical dicts; the ``__metadata`` wrapper is added here. SF validates the batch as a whole, so a multi-record batch is all-or-nothing. Returns the per-record envelopes in order; the caller decides success (SF answers HTTP 200 even for record-level failures — the verdict is ``status`` / ``httpCode`` / ``message`` inside each envelope). """ payload = [ {"__metadata": {"uri": entity}, **rec} for rec in records ] body = await self._post_json( "upsert", json_body=payload, params={"$format": "json"}, ) # Envelope variants seen across SF tenants: {"d": [records]} and # a bare [records]. Normalise to a list of record dicts. out = body.get("d") if isinstance(body, dict) else body if isinstance(out, dict): out = [out] return [r for r in out if isinstance(r, dict)] if isinstance(out, list) else [] async def list_employments_modified_since( self, since: datetime, ) -> list[dict[str, Any]]: """Employment records modified since the given timestamp. Pairs with :meth:`list_users_modified_since` so downstream lifecycle-event creation can distinguish onboard (startDate) from offboard (endDate) using the employment row's dates. """ return await self._list_entities( "EmpEmployment", select_fields=_EMP_EMPLOYMENT_SELECT_FIELDS, filter_clause=( f"lastModifiedDateTime ge { _odata_datetime_literal(since) }" ), ) # ── EC Time Off (annual leave) ──────────────────────────────── # # SF EC Time-Off is a separately-licensed module. The entity names # and time-account-type codes are tenant-configured — what works on # one PingCAP SF instance may need adjustment on another. Calls below # are defensive: on 404 / 401 / 403 (module not enabled or auth # scope missing) they return an empty list and log a warning rather # than raising. The 年假计算器 page then falls back to manual HR # input, which is its primary UX anyway. # # Entity names (PingCAP CN tenant, ``api30.sapsf.cn``, verified # via $metadata + sample calls 2026-05-14): # ``TimeAccount`` — account ledger per (user, accountType, # fiscal-year). Carries startDate / # endDate / bookingStart / bookingEnd. # Supports OR-batched userId filters. # ``TimeAccountDetail`` — accrual + booking line items per # account, FK ``TimeAccount_externalCode``. # Supports OR-batched UUID filters. # ``bookingType`` ∈ {ACCRUAL, # MANUAL_ADJUSTMENT, EMPLOYEE_TIME, # DEDUCTION, …}. # ``EmpTimeAccountBalance`` — current balance per (user, type); # DOES NOT accept ``or`` in $filter # (tenant restriction # ``COE_HRIS_QUERY_USING_OR_IN_FILTER``), # so it's only usable single-user. # We derive balance from # TimeAccountDetail aggregation instead # (sum of bookingAmount per account) # so the carry-forward sync stays batched. # # The ``CHN_AnnualLeave`` / ``CHN_PL`` codes below are the China-CN # tenant defaults (法定年假 + 福利年假) — different tenants must # override via constructor args on the caller side, not by editing # this file. Pre-2026-05-14 code targeted the non-existent # ``EmpTimeAccount`` entity and 404'd silently; if you're seeing # the legacy method name in any caller, replace with # :meth:`list_time_accounts` + :meth:`list_time_account_details`. async def list_time_accounts( self, *, user_ids: list[str], account_types: list[str] | None = None, ) -> list[dict[str, Any]]: """Per-(user, accountType, fiscal-year) account-ledger rows. ``TimeAccount`` returns one row per fiscal-year leave bucket. Carries ``externalCode`` (the UUID downstream callers join on), ``userId``, ``accountType``, ``startDate``, ``endDate``, ``bookingStartDate`` (typically ``startDate``), ``bookingEndDate`` (allows post-FY grace-period bookings), and ``accountClosed``. ``account_types``: * a non-empty list → restrict to those accountType codes (the legacy carryover path: {法定年假, 福利年假}). * ``None`` / empty → **no type filter**, returning *every* account type the user has (法定年假 / 福利年假 / 带薪病假 / 调休假 / …). This is the data-driven full-mirror path used by the leave-domain sync — we don't want to enumerate tenant-specific codes there. Returns ``[]`` (with a warning log) on 404 / auth failure so the caller can fall back to manual input rather than fail. """ if not user_ids: return [] # account-type clause is shared across chunks (omitted when the # caller wants every type). type_clause = "" if account_types: type_clause = " or ".join( f"accountType eq '{_odata_str(t)}'" for t in account_types ) results: list[dict[str, Any]] = [] for chunk in _chunked(user_ids, size=50): user_clause = " or ".join( f"userId eq '{_odata_str(u)}'" for u in chunk ) filter_clause = ( f"({user_clause}) and ({type_clause})" if type_clause else f"({user_clause})" ) try: chunk_results = await self._list_entities( "TimeAccount", select_fields=[ "externalCode", "userId", "accountType", "startDate", "endDate", "bookingStartDate", "bookingEndDate", "accountClosed", ], filter_clause=filter_clause, ) results.extend(chunk_results) except Exception as exc: # noqa: BLE001 — tenant-config drift import logging logging.getLogger(__name__).warning( "sf.time_accounts fetch failed for chunk of %d users " "(%s). 年假计算器 will fall back to HR manual input.", len(chunk), exc, ) # One bad chunk shouldn't tank the rest of the sync. continue return results async def list_time_account_details( self, *, time_account_uuids: list[str], ) -> list[dict[str, Any]]: """Per-account accrual + booking line items. Filtered by ``TimeAccount_externalCode`` (the UUID returned by :meth:`list_time_accounts`). Each row carries ``bookingType`` (ACCRUAL / MANUAL_ADJUSTMENT / EMPLOYEE_TIME / DEDUCTION …), ``bookingAmount`` (string-typed Decimal), ``bookingUnit``, ``bookingDate``, ``accrualPeriodId``, ``createdBy``, ``createdDateTime``. Used by the carry-forward sync to derive both balance (= Σ bookingAmount per account, sign-aware) and used-days (= Σ |bookingAmount| of leave-taken types like EMPLOYEE_TIME / DEDUCTION) — see ``hr.service.sync_annual_leave_carryover``. OR-joined batching is supported by this entity (verified 2026-05-14 on api30.sapsf.cn). """ if not time_account_uuids: return [] # Core fields the carry-forward math needs — proven on the CN # tenant (2026-05-14). NEVER fail the chunk over these. core_fields = [ "externalCode", "TimeAccount_externalCode", "bookingType", "bookingAmount", "bookingUnit", "bookingDate", "accrualPeriodId", ] # Extra audit fields surfaced in SF's 账户假期记录 grid (创建者 / # 创建日期 / 评论 / 发布日期). These are standard EC audit columns # but a tenant that doesn't expose one would 400 the whole # ``$select``, so we try the enriched select first and fall back # to ``core_fields`` per chunk — the leave ledger still loads, # just without the optional columns. rich_fields = core_fields + [ "createdBy", "createdDateTime", "comment", "postingDate", ] import logging _log = logging.getLogger(__name__) results: list[dict[str, Any]] = [] # 16 uuids/call keeps the encoded filter under tenant URL limits # while still cutting round-trips by ~16× vs single-uuid calls. for chunk in _chunked(time_account_uuids, size=16): uuid_clause = " or ".join( f"TimeAccount_externalCode eq '{_odata_str(u)}'" for u in chunk ) try: chunk_results = await self._list_entities( "TimeAccountDetail", select_fields=rich_fields, filter_clause=uuid_clause, ) results.extend(chunk_results) except Exception as exc: # noqa: BLE001 — maybe a missing field # Retry with the minimal proven select before giving up, # so an unsupported audit column never costs us the ledger. try: chunk_results = await self._list_entities( "TimeAccountDetail", select_fields=core_fields, filter_clause=uuid_clause, ) results.extend(chunk_results) _log.info( "sf.time_account_details: enriched $select rejected " "(%s); fell back to core fields for chunk of %d.", exc, len(chunk), ) except Exception as exc2: # noqa: BLE001 — tenant-config drift _log.warning( "sf.time_account_details fetch failed for chunk of %d " "uuids (%s). Their carry-forward will fall back to HR " "manual input.", len(chunk), exc2, ) continue return results async def list_time_account_types(self) -> dict[str, str]: """Map ``TimeAccountType.externalCode`` → human label (名称). Resolves the localized account-type names SF shows in its 「假期余额」 cards (法定年假 / 福利年假 / 带薪病假 / 调休假) so the Helix mirror can store a display label next to each account's ``accountType`` code. Best-effort: ``TimeAccountType`` is part of the same EC Time-Off module and may be unlicensed / unexposed on a given tenant, so a 404 / auth / unknown-field error returns ``{}`` and the caller falls back to a code→i18n map. Selects only the two universally- present fields so a tenant-specific localized column can't 400 the call. """ try: rows = await self._list_entities( "TimeAccountType", select_fields=["externalCode", "name"], ) except Exception as exc: # noqa: BLE001 — module/field may be absent import logging logging.getLogger(__name__).info( "sf.time_account_types unavailable (%s); leave-account " "labels fall back to the code→i18n map.", exc, ) return {} out: dict[str, str] = {} for row in rows: code = row.get("externalCode") name = row.get("name") if code and name: out[str(code)] = str(name) return out async def list_employee_time( self, *, user_ids: list[str], ) -> list[dict[str, Any]]: """Per-user time-off REQUESTS (``EmployeeTime``) — the source of the leave *reason* / *type* / *approval*, which ``TimeAccountDetail`` does NOT carry (that entity only records the accounting deduction). Each row carries ``userId``, ``startDate`` / ``endDate`` (the leave span), ``timeType`` (absence-type code), ``quantityInDays``, ``approvalStatus`` and ``comment`` (the free-text 事由, when the employee entered one). The leave-data sync matches these back onto the deduction postings by (userId, booking_date ∈ [start, end]). Best-effort like the other EC Time-Off helpers: a tenant without the module / a missing field returns ``[]`` (enriched-select rejection falls back to the core fields first) so leave records simply show no reason rather than failing the sync. OR-batched on ``userId``. """ if not user_ids: return [] core_fields = [ "externalCode", "userId", "startDate", "endDate", "timeType", "quantityInDays", ] rich_fields = core_fields + ["approvalStatus", "comment"] import logging _log = logging.getLogger(__name__) results: list[dict[str, Any]] = [] for chunk in _chunked(user_ids, size=50): user_clause = " or ".join( f"userId eq '{_odata_str(u)}'" for u in chunk ) try: results.extend(await self._list_entities( "EmployeeTime", select_fields=rich_fields, filter_clause=user_clause, )) except Exception as exc: # noqa: BLE001 — maybe missing field try: results.extend(await self._list_entities( "EmployeeTime", select_fields=core_fields, filter_clause=user_clause, )) _log.info( "sf.employee_time: enriched $select rejected (%s); " "fell back to core fields for chunk of %d.", exc, len(chunk), ) except Exception as exc2: # noqa: BLE001 — module/auth absent _log.warning( "sf.employee_time fetch failed for chunk of %d users " "(%s). Leave records will show no reason/type.", len(chunk), exc2, ) continue return results # ── Diagnostics ─────────────────────────────────────────────── async def ping(self) -> dict[str, Any]: """Lightweight connectivity test. Issues one ``FOCompany?$top=1`` call — smallest-possible request that exercises both the base URL and the configured auth. Returns a dict summarizing success; exceptions bubble up to the caller. """ results = await self._list_entities( "FOCompany", select_fields=["externalCode", "name"], top=1, ) return { "ok": True, "base_url": self.base_url, "auth": self.auth.describe(), "sample_company": (results or [{}])[0].get("name"), } def _chunked(values: list[str], *, size: int) -> Iterable[list[str]]: for i in range(0, len(values), size): yield values[i:i + size] def _odata_str(value: str) -> str: """Escape a string for interpolation inside an OData v2 single-quoted literal (``'`` → ``''``). Every ``... eq '{value}'`` interpolation MUST route through this — review round-1 2026-07-04 found three sites building filters from raw ids while ``list_time_accounts`` already escaped; a quote in an SF userId would otherwise break the whole query. """ return value.replace("'", "''") def _odata_datetime_literal(value: date | datetime) -> str: if isinstance(value, datetime): dt = value else: dt = datetime.combine(value, datetime.min.time()) return f"datetime'{dt:%Y-%m-%dT%H:%M:%S}'" def _effective_asof_params(effective_as_of: date | None) -> dict[str, str] | None: """Build the SF EC effective-dating query params that pin an effective-dated entity read to the slice effective on a single date. SuccessFactors EmpJob / EmpJobRelationships are effective-dated; without a date param SF returns the as-of-today slice, which for a future hire is empty (their row is future-dated). ``fromDate == toDate == `` asks SF for the slice effective on exactly that date, so the 24h-lead build can read a pre-start hire's future-dated department / manager / HRBP / HR Ops. Format is ``YYYY-MM-DD`` (SF EC navigation params, not ``$``-OData). Returns ``None`` when no date is given so the caller's query stays unchanged. """ if effective_as_of is None: return None iso = effective_as_of.isoformat() return {"fromDate": iso, "toDate": iso}