From fc5da5fe3ada131b3e0d0bd78452f3fe0d0ae211 Mon Sep 17 00:00:00 2001 From: Nithin Date: Sun, 30 Aug 2026 06:04:38 -0400 Subject: [PATCH] feat(workday): shard capped public job boards --- src/fortune_intel/connectors/workday.py | 141 +++++++++++++++++++++++- tests/test_workday_connector.py | 73 ++++++++++++ 2 files changed, 213 insertions(+), 1 deletion(-) diff --git a/src/fortune_intel/connectors/workday.py b/src/fortune_intel/connectors/workday.py index eb38db0..1c3aae9 100644 --- a/src/fortune_intel/connectors/workday.py +++ b/src/fortune_intel/connectors/workday.py @@ -114,6 +114,8 @@ def __init__( detail_concurrency: int = 4, detail_client_factory: Callable[[], JsonHttpClient] | None = None, reported_total_probe_threshold: int = _REPORTED_TOTAL_PROBE_THRESHOLD, + _applied_facets: dict[str, list[str]] | None = None, + _allow_facet_sharding: bool = True, ) -> None: self.workday = parse_workday_source_key(source_key) if not 1 <= page_size <= 20: @@ -127,6 +129,11 @@ def __init__( if not 1 <= detail_concurrency <= 8: raise ValueError("detail_concurrency must be between 1 and 8") self.detail_concurrency = detail_concurrency + self._applied_facets = { + str(parameter): [str(value) for value in values] + for parameter, values in (_applied_facets or {}).items() + } + self._allow_facet_sharding = _allow_facet_sharding self._detail_client_factory = detail_client_factory or ( JsonHttpClient if client is None else lambda: self.client ) @@ -142,6 +149,7 @@ def fetch(self) -> ConnectorResult: pages = 0 expected_total: int | None = None reported_total_was_truncated = False + seen_page_signatures: set[tuple[str, ...]] = set() list_url = f"{self.workday.cxs_base_url}/jobs" while pages < self.max_pages: @@ -150,7 +158,7 @@ def fetch(self) -> ConnectorResult: payload = self.client.post_json( list_url, json_body={ - "appliedFacets": {}, + "appliedFacets": self._applied_facets, "limit": self.page_size, "offset": offset, "searchText": "", @@ -180,6 +188,31 @@ def fetch(self) -> ConnectorResult: errors.append(record_error(error)) return self._result(jobs, errors, pages) + if pages == 1 and self._allow_facet_sharding: + facet_shards = self._facet_shards(payload, total) + if facet_shards is not None: + parameter, values = facet_shards + return self._fetch_faceted_manifest(parameter, values) + + # Some public Workday tenants cap the reported total and reset the + # offset back to page one after that cap. Continuing would request + # the same jobs hundreds of times and falsely look like a slow + # board. A repeated page is never a complete manifest, so fail + # fast while preserving the previously successful job snapshot. + page_signature = self._page_signature(postings) + if postings and page_signature in seen_page_signatures: + errors.append( + record_error( + ValueError( + "pagination repeated a listing page; public board offset " + "is capped or reset" + ) + ) + ) + return self._result(jobs, errors, pages) + if postings: + seen_page_signatures.add(page_signature) + if expected_total is None: expected_total = total elif total not in {0, expected_total}: @@ -276,6 +309,112 @@ def _parse_page(payload: object) -> tuple[int, list[dict[str, object]]]: raise ValueError("job-list total must be a non-negative integer") return total, postings + def _fetch_faceted_manifest( + self, + parameter: str, + values: tuple[str, ...], + ) -> ConnectorResult: + """Fetch a capped public board through exclusive Workday facet shards.""" + + jobs: list[ConnectorJob] = [] + errors: list[ConnectorError] = [] + seen_ids: set[str] = set() + pages = 1 # The root response discovered the exact public facets. + for value in values: + result = WorkdayConnector( + self.workday.key, + page_size=self.page_size, + max_pages=self.max_pages, + client=self.client, + detail_concurrency=self.detail_concurrency, + detail_client_factory=self._detail_client_factory, + reported_total_probe_threshold=self.reported_total_probe_threshold, + _applied_facets={parameter: [value]}, + _allow_facet_sharding=False, + ).fetch() + pages += result.pages_fetched + if not result.complete: + errors.extend(result.errors) + continue + for job in result.jobs: + if job.external_job_id in seen_ids: + errors.append( + record_error( + ValueError( + "a Workday job appeared in multiple facet shards; " + "manifest cannot be proven complete" + ), + external_id=job.external_job_id, + ) + ) + continue + seen_ids.add(job.external_job_id) + jobs.append(job) + return ConnectorResult( + self.source, + self.workday.key, + tuple(jobs), + not errors, + tuple(errors), + pages, + ) + + def _facet_shards(self, payload: object, total: int) -> tuple[str, tuple[str, ...]] | None: + """Choose observed finite facet buckets only when a root total is capped.""" + + if total < self.reported_total_probe_threshold or not isinstance(payload, dict): + return None + facets = payload.get("facets") + if not isinstance(facets, list): + return None + for facet in facets: + if not isinstance(facet, dict): + continue + parameter = clean_text(facet.get("facetParameter")) + raw_values = facet.get("values") + if not parameter or not isinstance(raw_values, list): + continue + values: list[tuple[str, int]] = [] + for item in raw_values: + if not isinstance(item, dict): + values = [] + break + identifier = clean_text(item.get("id")) + try: + count = int(item.get("count")) + except (TypeError, ValueError): + values = [] + break + if not identifier or count < 1 or count >= self.reported_total_probe_threshold: + values = [] + break + values.append((identifier, count)) + # More than one sub-cap bucket and an aggregate beyond the root + # total show that the anonymous root listing is capped. We submit + # only its observed public facet IDs back to the same endpoint. + if 2 <= len(values) <= 100 and sum(count for _, count in values) > total: + return parameter, tuple(identifier for identifier, _ in values) + return None + + @staticmethod + def _page_signature(postings: list[dict[str, object]]) -> tuple[str, ...]: + """Fingerprint one listing page without trusting an unbounded payload.""" + + return tuple( + "\x1f".join( + ( + clean_text(posting.get("externalPath")), + clean_text(posting.get("title")), + "\x1e".join( + clean_text(value) + for value in posting.get("bulletFields", []) + if isinstance(value, str) + ), + ) + ) + for posting in postings + ) + @staticmethod def _detail_path(value: str) -> str: parsed = urlsplit(value) diff --git a/tests/test_workday_connector.py b/tests/test_workday_connector.py index d49ba3a..d813d7e 100644 --- a/tests/test_workday_connector.py +++ b/tests/test_workday_connector.py @@ -326,6 +326,79 @@ def test_marks_manifest_incomplete_if_total_changes_during_pagination(): assert "total changed" in result.errors[0].message +def test_stops_when_public_workday_listing_repeats_a_page_after_total_cap(): + """A capped board can reset offsets; never spin until the generic page limit.""" + + first = workday_summary("JR-1") + second = workday_summary("JR-2") + client = StubClient( + [ + {"total": 2, "jobPostings": [first]}, + {"total": 0, "jobPostings": [second]}, + {"total": 2, "jobPostings": [first]}, + ] + ) + key = workday_source("acme.wd5.myworkdayjobs.com", "acme", "External").key + + result = WorkdayConnector( + key, + page_size=1, + reported_total_probe_threshold=2, + client=client, + ).fetch() + + assert result.complete is False + assert result.jobs == () + assert result.pages_fetched == 3 + assert "repeated a listing page" in result.errors[0].message + + +def test_uses_observed_workday_facets_to_complete_a_capped_public_board(): + root = { + "total": 2, + "jobPostings": [workday_summary("ROOT")], + "facets": [ + { + "facetParameter": "jobFamilyGroup", + "values": [ + {"id": "engineering", "count": 1}, + {"id": "sales", "count": 1}, + {"id": "marketing", "count": 1}, + ], + } + ], + } + client = StubClient( + [ + root, + {"total": 1, "jobPostings": [workday_summary("JR-1")]}, + workday_detail("JR-1"), + {"total": 1, "jobPostings": [workday_summary("JR-2")]}, + workday_detail("JR-2"), + {"total": 1, "jobPostings": [workday_summary("JR-3")]}, + workday_detail("JR-3"), + ] + ) + key = workday_source("acme.wd5.myworkdayjobs.com", "acme", "External").key + + result = WorkdayConnector( + key, + reported_total_probe_threshold=2, + client=client, + ).fetch() + + assert result.complete is True + assert result.pages_fetched == 4 + assert [job.external_job_id for job in result.jobs] == [ + "opaque-JR-1", + "opaque-JR-2", + "opaque-JR-3", + ] + assert client.calls[1][1]["appliedFacets"] == {"jobFamilyGroup": ["engineering"]} + assert client.calls[3][1]["appliedFacets"] == {"jobFamilyGroup": ["sales"]} + assert client.calls[5][1]["appliedFacets"] == {"jobFamilyGroup": ["marketing"]} + + def test_accepts_zero_total_sentinel_after_first_page(): client = StubClient( [