"""Person ↔ GitHub handle lookup — the people surface of the External API. POST /api/v1/open/people/github/lookup people.github:read The point-query twin of the Directory pull API's ``github_id`` field. The Directory plane (``helix_svc_`` credentials) exists for systems that *synchronise the roster* — keyset incremental, field-level grants. This endpoint exists for the other half of the job: a caller that already has a question and wants an answer now, without standing up a sync. Three selectors, combinable in one call, each answering a real direction of the GitHub-reconciliation problem: * ``emails`` — "what is this person's handle?" (forward) * ``github_ids`` — "who is this handle, and are they still employed?" (**reverse** — the direction a GitHub org member list actually gives you, and the one a lookup-by-email API cannot answer) * ``departments`` — "everyone under this org path" (the roster slice a team-level audit needs) Design contract (documented for integrators in ``docs/api/external-api.md`` §B.7): * **Population ceiling** — ``departments`` makes this the first External API surface that LISTS people, so unlike the lookup-only leave API it cannot lean on "you must already know who to ask about" as its boundary. Every request is intersected with the client's ``person_visibility`` (see :mod:`app.api_clients.person_visibility`); a person outside it is reported exactly like a person who does not exist, so the surface is never an existence oracle. * **Explicit tri-state** — a requested email comes back ``found`` / ``no_github_id`` / ``unknown_person``. "We have this person but no handle on file" must never collapse into "no such person": the first is a coverage gap to chase, the second is a stale directory. Same reasoning as the leave API's ``coverage`` tri-state. * **Sparse by nature** — ``github_id`` is collected at onboarding and never backfilled, so ``null`` means "not on file", NOT "has no GitHub account". A caller reconciling org membership must never read a null as "not an employee" and remove them on that basis; the actionable signal is the reverse (a handle in the org that Helix cannot name). * **Employment scope** — the default population is people who are currently employed (``active`` / ``leaver_pending``; someone serving notice still holds access, so an access audit still wants them). ``include_left=true`` adds departed staff, which is what an offboarding audit needs. ``pre_hire`` is NEVER exposed here: an unannounced future hire has no GitHub access to reconcile and their existence is not this surface's business. * **Egress parity** — HRIS-purged rows, SF no-push opt-outs (``2061``) and non-corporate-email junk records are excluded, exactly as on the Directory pull surface. Both planes must agree on who may egress. """ from __future__ import annotations import logging from typing import Any from fastapi import APIRouter, Depends, Request from pydantic import BaseModel, Field, field_validator from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.api_clients.auth import ApiClientContext, require_api_client from app.api_clients.person_visibility import person_visibility_is_unrestricted from app.api_clients.scopes import SCOPE_PEOPLE_GITHUB_READ from app.core.database import get_db from app.core.exceptions import ValidationError from app.core.response import success_response from app.identity.bu_helpers import department_segments from app.identity.consumers.directory_service import resolve_in_scope_person_ids from app.identity.person_models import IdentityOrgMembership, IdentityPerson logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/open/people", tags=["open-people"]) _MAX_EMAILS = 200 _MAX_GITHUB_IDS = 200 _MAX_DEPARTMENTS = 50 #: Refuse rather than truncate. A ~500-person company cannot legitimately #: exceed this from a department query; if it ever does, silently cutting #: the list would read as "that's the whole team" to a reconciler. _MAX_PEOPLE = 1000 #: Employment statuses that mean "currently employed" — the default #: population. ``leaver_pending`` is in notice period: still staff, still #: holding access, therefore still in scope for an access audit. #: #: Deliberately mirrors ``IdentityPerson._ALLOWED_EMPLOYMENT_STATUSES`` #: minus the departed/not-yet-arrived ends. There is no ``on_leave`` #: value in this platform (the model's @validates hook rejects it), so #: listing one here — and promising it in the docs — would describe a #: population that cannot exist. _EMPLOYED_STATUSES = frozenset({"active", "leaver_pending"}) #: Added by ``include_left``. _DEPARTED_STATUSES = frozenset({"left"}) _LOOKUP_EXAMPLE = { "success": True, "data": { "emails": [ { "email": "zhangsan@pingcap.com", "status": "found", "person": { "work_email": "zhangsan@pingcap.com", "display_name": "Zhang San", "employee_no": "P12345", "github_id": "zhangsan-pc", "employment_status": "active", "department_full_path": "Engineering Group / TiDB", }, }, { "email": "nohandle@pingcap.com", "status": "no_github_id", "person": { "work_email": "nohandle@pingcap.com", "display_name": "No Handle", "employee_no": "P12346", "github_id": None, "employment_status": "active", "department_full_path": "Engineering Group / TiKV", }, }, {"email": "ghost@pingcap.com", "status": "unknown_person", "person": None}, ], "github_ids": [ {"github_id": "octo-han", "status": "found", "person": {"…": "…"}}, {"github_id": "not-ours", "status": "unknown_handle", "person": None}, ], "departments": { "requested": ["Engineering Group / TiDB"], "unmatched": [], "people": [{"…": "…"}], }, }, "meta": { "trace_id": "…", "visibility": "unrestricted", "include_left": False, "person_count": 12, }, "errors": [], } class GithubLookupRequest(BaseModel): """Body for the person ↔ GitHub handle lookup. At least one selector is required — an empty body would mean "the whole roster", which is never what a caller wants to ask for and exactly what the population ceiling exists to prevent. """ emails: list[str] = Field(default_factory=list, max_length=_MAX_EMAILS) github_ids: list[str] = Field(default_factory=list, max_length=_MAX_GITHUB_IDS) departments: list[str] = Field( default_factory=list, max_length=_MAX_DEPARTMENTS, ) include_left: bool = False @field_validator("emails") @classmethod def _clean_emails(cls, v: list[str]) -> list[str]: return _dedupe_lower(v) @field_validator("github_ids") @classmethod def _clean_handles(cls, v: list[str]) -> list[str]: # GitHub handles are case-insensitive; compare and de-duplicate # folded, but the response echoes what the caller sent so they can # join the answer back onto their own list. return _dedupe_preserving_case(v) @field_validator("departments") @classmethod def _clean_departments(cls, v: list[str]) -> list[str]: return _dedupe_preserving_case(v) def _dedupe_lower(values: list[str]) -> list[str]: out: list[str] = [] seen: set[str] = set() for raw in values: cleaned = (raw or "").strip().lower() if cleaned and cleaned not in seen: seen.add(cleaned) out.append(cleaned) return out def _dedupe_preserving_case(values: list[str]) -> list[str]: out: list[str] = [] seen: set[str] = set() for raw in values: cleaned = (raw or "").strip() key = cleaned.lower() if cleaned and key not in seen: seen.add(key) out.append(cleaned) return out def _clean_handle(raw: str | None) -> str | None: """A stored GitHub handle, trimmed; ``None`` when there is nothing usable left. Whitespace-only is "not on file", not a handle.""" cleaned = (raw or "").strip() return cleaned or None def _normalize_path(value: str) -> str: """Canonical comparison form for a department path. Splits on ``/``, drops empty segments and case, then re-joins — so ``"engineering group/tidb"``, ``"Engineering Group / TiDB"`` and ``" Engineering Group /TiDB "`` are one and the same request. The tenant root is stripped through the same helper the person side uses, so a caller who copies a path out of another Helix surface (where paths read ``"PingCAP / Engineering"``) is understood rather than silently told the department does not exist. """ segments = department_segments(value) return "/".join(seg.lower() for seg in segments) def _departments_block( body: GithubLookupRequest, matched_keys: set[str], people: list[dict], ) -> dict: """The ``departments`` response block, built one way for both the normal and the nobody-in-scope path. ``unmatched`` is derived from what the CALLER sent, filtered by the normalized keys that matched — not from the normalized map. Two spellings of the same path (``"A / B"`` and ``"A/B"``) collapse to one key but are two requested entries, so deriving it from the map would silently drop one of them from the caller's own echo. """ return { "requested": body.departments, # A path that matched no department at all — a typo, or a # department outside this client's population ceiling. Told # explicitly so "empty result" is never ambiguous with # "wrong path". "unmatched": [ d for d in body.departments if _normalize_path(d) not in matched_keys ], "people": people, } def _serialize(person: IdentityPerson, org_path: str | None) -> dict: """The wire shape of one person on this surface. Deliberately narrow: identity + handle + the department path, and the employment status so a reconciler can act on leavers. No contact details — this scope is a mapping table, not a directory export. """ segments = department_segments(org_path) return { "work_email": person.work_email, "display_name": person.display_name, "employee_no": person.employee_no, # Stripped: a stored handle with stray whitespace is unusable on # GitHub, and echoing it raw would hand the caller a value that # silently fails to match anything on their side. "github_id": _clean_handle(person.github_id), "employment_status": person.employment_status, "department_full_path": " / ".join(segments) if segments else None, } @router.post( "/github/lookup", summary="Look up the person ↔ GitHub handle mapping", responses={200: {"content": {"application/json": {"example": _LOOKUP_EXAMPLE}}}}, ) async def lookup_github_ids( body: GithubLookupRequest, request: Request, ctx: ApiClientContext = Depends(require_api_client(SCOPE_PEOPLE_GITHUB_READ)), db: AsyncSession = Depends(get_db), ) -> dict: if not (body.emails or body.github_ids or body.departments): raise ValidationError( "Provide at least one of `emails`, `github_ids` or `departments`.", ) # ── Population ceiling ──────────────────────────────────────────── # A stored ceiling that isn't a JSON object cannot be evaluated, and # "cannot evaluate" must mean nobody — never the ``None`` meaning of # "unrestricted". Checked before anything else so no later branch can # reinstate whole-roster reach. if ctx.person_visibility_malformed: logger.warning( "external-api github lookup: client=%s has an unparseable " "person_visibility; failing closed", ctx.slug, ) return _empty_response(request, body, unrestricted=False) # The client's person_visibility bounds everything below. Unrestricted # clients get the whole roster MINUS the egress gates, which is what # `include_offboarded=True` on an otherwise-empty filter means: it # disables the engine's default active-only pre-gate, leaving the # status decision to this endpoint's own `include_left` contract # rather than silently applying a second, invisible one. # # The same override is applied to a RESTRICTED ceiling that names no # employment status. Without it the engine's active-only pre-gate # would silently drop leaver_pending people from, say, a # departments-only ceiling — contradicting this endpoint's documented # default population, and hiding exactly the notice-period staff an # access audit is looking for. A ceiling that DOES name statuses is # passed through untouched: that is the admin's explicit choice, and # ``include_offboarded`` would widen it by adding ``left``. unrestricted = person_visibility_is_unrestricted(ctx.person_visibility) ceiling = dict(ctx.person_visibility or {}) if not ceiling.get("employment_statuses"): ceiling["include_offboarded"] = True in_scope_ids = await resolve_in_scope_person_ids(db, ceiling) if not in_scope_ids: # A restricted client whose filter matches nobody sees nothing — # never degraded into "unrestricted" (fail-closed). return _empty_response(request, body, unrestricted) allowed_statuses = set(_EMPLOYED_STATUSES) if body.include_left: allowed_statuses |= _DEPARTED_STATUSES persons = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(in_scope_ids), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status.in_(sorted(allowed_statuses)), ) )).scalars().all() # Primary org membership → department path, in one pass. org_path_by_person: dict[str, str | None] = {} if persons: rows = (await db.execute( select( IdentityOrgMembership.person_id, IdentityOrgMembership.org_path, ).where( IdentityOrgMembership.person_id.in_([str(p.id) for p in persons]), IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) )).all() org_path_by_person = {str(r.person_id): r.org_path for r in rows} by_email: dict[str, IdentityPerson] = {} by_handle: dict[str, IdentityPerson] = {} # Stable iteration order so a collision (two rows sharing an email or # a handle — a data-quality artefact that does occur) always resolves # the same way. Without this the winner is whatever order the DB # happened to return, and the same request could answer differently on # two consecutive calls, which is indistinguishable from the roster # having changed. for p in sorted(persons, key=lambda p: (str(p.employee_no or ""), str(p.id))): email_key = (p.work_email or "").strip().lower() if email_key: # Two rows sharing an email: prefer the employed one so the # answer describes the live record. current = by_email.get(email_key) if current is None or ( p.employment_status in _EMPLOYED_STATUSES and current.employment_status not in _EMPLOYED_STATUSES ): by_email[email_key] = p handle_key = (p.github_id or "").strip().lower() if handle_key: current_h = by_handle.get(handle_key) if current_h is None or ( p.employment_status in _EMPLOYED_STATUSES and current_h.employment_status not in _EMPLOYED_STATUSES ): by_handle[handle_key] = p data: dict = {} returned: set[str] = set() def _person_payload(p: IdentityPerson) -> dict: returned.add(str(p.id)) return _serialize(p, org_path_by_person.get(str(p.id))) # Presence keys off what the caller SENT, not off what survived # cleaning: ``{"emails": [" "], "departments": [...]}`` cleans the # emails to nothing, and dropping the key would force a client to # special-case a missing block — the exact thing ``_empty_response`` # exists to prevent. if "emails" in body.model_fields_set: # Annotated: the not-found branch appends first, so an inferred # type would lock the rows to ``str | None`` and reject the # nested person object on the found branch. email_rows: list[dict[str, Any]] = [] for email in body.emails: person = by_email.get(email) if person is None: # Out-of-ceiling, out-of-status, gated, or genuinely # absent — all one answer, on purpose. email_rows.append({ "email": email, "status": "unknown_person", "person": None, }) continue email_rows.append({ "email": email, # Cleaned, so a whitespace-only stored handle reports the # coverage gap it actually is instead of a `found` whose # payload carries an unusable value. "status": ( "found" if _clean_handle(person.github_id) else "no_github_id" ), "person": _person_payload(person), }) data["emails"] = email_rows if "github_ids" in body.model_fields_set: handle_rows: list[dict[str, Any]] = [] for handle in body.github_ids: person = by_handle.get(handle.lower()) if person is None: handle_rows.append({ "github_id": handle, "status": "unknown_handle", "person": None, }) continue handle_rows.append({ "github_id": handle, "status": "found", "person": _person_payload(person), }) data["github_ids"] = handle_rows if "departments" in body.model_fields_set: # Prefixed once, not per person: the inner loop runs # len(departments) × len(roster) times. wanted = [(_normalize_path(d), _normalize_path(d) + "/") for d in body.departments] matched_keys: set[str] = set() dept_people: list[dict] = [] # Same stability argument as the lookup maps above: a caller that # diffs two pulls of the same department should see real changes, # not row-order churn. for p in sorted(persons, key=lambda p: (str(p.work_email or ""), str(p.id))): segments = department_segments(org_path_by_person.get(str(p.id))) if not segments: continue person_path = "/".join(s.lower() for s in segments) hit = False for key, key_prefix in wanted: # Subtree semantics: a requested path matches itself and # everything below it. The trailing "/" guard stops # "Engineering" from swallowing "Engineering Support". if key and (person_path == key or person_path.startswith(key_prefix)): # EVERY matching key is recorded, not just the first: # asking for both a parent and its child must not # report the child as unmatched merely because the # parent matched this person first. matched_keys.add(key) hit = True if hit: dept_people.append(_person_payload(p)) if len(dept_people) > _MAX_PEOPLE: raise ValidationError( f"department selection matched {len(dept_people)} people " f"(limit {_MAX_PEOPLE}); narrow the department paths.", ) data["departments"] = _departments_block(body, matched_keys, dept_people) logger.info( "external-api github lookup client=%s emails=%d handles=%d depts=%d " "left=%s visibility=%s people=%d", ctx.slug, len(body.emails), len(body.github_ids), len(body.departments), body.include_left, "unrestricted" if unrestricted else "restricted", len(returned), ) return success_response( data=data, request=request, meta={ "visibility": "unrestricted" if unrestricted else "restricted", "include_left": body.include_left, "person_count": len(returned), }, ) def _empty_response( request: Request, body: GithubLookupRequest, unrestricted: bool, ) -> dict: """Shape-stable answer when the ceiling resolves to nobody. Every selector the caller sent still comes back, with each entry in its not-found state — a client parsing ``data["emails"]`` must not have to special-case a missing key just because its grant is narrow. """ data: dict = {} if "emails" in body.model_fields_set: data["emails"] = [ {"email": e, "status": "unknown_person", "person": None} for e in body.emails ] if "github_ids" in body.model_fields_set: data["github_ids"] = [ {"github_id": h, "status": "unknown_handle", "person": None} for h in body.github_ids ] if "departments" in body.model_fields_set: # Same builder as the normal path — nothing matched, so every # requested path is unmatched, derived by the one rule. data["departments"] = _departments_block(body, set(), []) return success_response( data=data, request=request, meta={ "visibility": "unrestricted" if unrestricted else "restricted", "include_left": body.include_left, "person_count": 0, }, ) __all__ = ["router"]