fix(ofsted): a gov.uk outage skips the FE streams, not the Ofsted extract (review)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/fe_mi.py

@@ -135,3 +135,54 @@ def records(df: pd.DataFrame, columns: dict[str, str], key: str) -> Iterator[dic
         record = {field: str(row[source]).strip() for source, field in present.items()}
         if record.get(key):
             yield record
+
+
+# requests' exceptions are OSErrors; a body that is not JSON is a ValueError.
+_UNAVAILABLE = (OSError, ValueError)
+
+
+def provider_records(
+    search: Callable[[], list[dict]],
+    fetch_page: Callable[[str], str],
+    fetch_csv: Callable[[str], bytes],
+    log: Callable[[str], None],
+) -> Iterator[dict]:
+    """The newest "as at" file's provider records. gov.uk being unreachable
+    yields nothing, logged: the FE streams share the monthly Ofsted extract
+    with the school MI and must not fail it."""
+    try:
+        url = providers_csv_url(fe_pages(search()), fetch_page)
+        if not url:
+            log("No Ofsted FE 'as at' file found")
+            return
+        df = read_mi_csv(fetch_csv(url))
+    except _UNAVAILABLE as exc:
+        log(f"Ofsted FE providers MI unavailable: {exc}")
+        return
+    yield from records(df, PROVIDER_COLUMNS, "provider_urn")
+
+
+def inspection_records(
+    search: Callable[[], list[dict]],
+    fetch_page: Callable[[str], str],
+    fetch_csv: Callable[[str], bytes],
+    log: Callable[[str], None],
+) -> Iterator[dict]:
+    """Renewed-framework inspection records from every page since November
+    2025. A page that cannot be read is logged and skipped; the others load."""
+    try:
+        pages = renewed_framework_pages(search())
+    except _UNAVAILABLE as exc:
+        log(f"Ofsted FE inspections MI unavailable: {exc}")
+        return
+    for page in pages:
+        try:
+            url = latest_period_csv(fetch_page(page))
+            if not url:
+                log(f"No renewed-framework FE inspections file on {page}")
+                continue
+            df = read_mi_csv(fetch_csv(url))
+        except _UNAVAILABLE as exc:
+            log(f"Ofsted FE inspections MI unavailable for {page}: {exc}")
+            continue
+        yield from records(df, INSPECTION_COLUMNS, "inspection_number")

pipeline/plugins/extractors/tap-uk-ofsted/tap_uk_ofsted/tap.py

@@ -15,12 +15,8 @@ from tap_uk_ofsted.fe_mi import (
     PROVIDER_COLUMNS,
     SEARCH_QUERY,
     SEARCH_URL,
-    fe_pages,
-    latest_period_csv,
-    providers_csv_url,
-    read_mi_csv,
-    records,
-    renewed_framework_pages,
+    inspection_records,
+    provider_records,
 )
 
 GOV_UK_PAGE = (
@@ -331,6 +327,12 @@ def _gov_uk_page(path: str) -> str:
     return resp.text
 
 
+def _gov_uk_file(url: str) -> bytes:
+    resp = requests.get(url, timeout=120)
+    resp.raise_for_status()
+    return resp.content
+
+
 class OfstedFEProvidersStream(Stream):
     """Ofsted's further education and skills providers: the newest "as at"
     file, one row per provider with its latest inspection under the previous
@@ -346,14 +348,7 @@ class OfstedFEProvidersStream(Stream):
     ).to_dict()
 
     def get_records(self, context):
-        url = providers_csv_url(fe_pages(_fe_search_results()), _gov_uk_page)
-        if not url:
-            self.logger.error("No Ofsted FE 'as at' file found")
-            return
-        self.logger.info("Downloading Ofsted FE providers MI: %s", url)
-        resp = requests.get(url, timeout=120)
-        resp.raise_for_status()
-        yield from records(read_mi_csv(resp.content), PROVIDER_COLUMNS, "provider_urn")
+        yield from provider_records(_fe_search_results, _gov_uk_page, _gov_uk_file, self.logger.error)
 
 
 class OfstedFEInspectionsStream(Stream):
@@ -371,15 +366,7 @@ class OfstedFEInspectionsStream(Stream):
     ).to_dict()
 
     def get_records(self, context):
-        for page in renewed_framework_pages(_fe_search_results()):
-            url = latest_period_csv(_gov_uk_page(page))
-            if not url:
-                self.logger.warning("No renewed-framework FE inspections file on %s", page)
-                continue
-            self.logger.info("Downloading Ofsted FE inspections MI: %s", url)
-            resp = requests.get(url, timeout=120)
-            resp.raise_for_status()
-            yield from records(read_mi_csv(resp.content), INSPECTION_COLUMNS, "inspection_number")
+        yield from inspection_records(_fe_search_results, _gov_uk_page, _gov_uk_file, self.logger.warning)
 
 
 class TapUKOfsted(Tap):