tudor/school_compare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(tap-uk-ees): handle plain list response from releases endpoint
All checks were successful
Build and Push Docker Images / Build Backend (FastAPI) (push) Successful in 33s
Details
Build and Push Docker Images / Build Frontend (Next.js) (push) Successful in 1m6s
Details
Build and Push Docker Images / Build Pipeline (Meltano + dbt + Airflow) (push) Successful in 1m45s
Details
Build and Push Docker Images / Trigger Portainer Update (push) Successful in 1s
Details

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Tudor Sitaru
2026-03-30 21:47:14 +01:00
parent 17617137ea
commit 570c2b689e
1 changed files with 6 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
pipeline/plugins/extractors/tap-uk-ees/tap_uk_ees/tap.py
View File

@@ -34,19 +34,12 @@ def get_content_release_id(publication_slug: str) -> str:
def get_all_release_ids(publication_slug: str) -> list[str]: def get_all_release_ids(publication_slug: str) -> list[str]:
"""Return all release IDs for a publication, newest first.""" """Return all release IDs for a publication, newest first."""
url = f"{CONTENT_API_BASE}/publications/{publication_slug}/releases" url = f"{CONTENT_API_BASE}/publications/{publication_slug}/releases"
ids = [] resp = requests.get(url, timeout=TIMEOUT)
page = 1 resp.raise_for_status()
while True: data = resp.json()
resp = requests.get(url, params={"page": page, "pageSize": 20}, timeout=TIMEOUT) # API returns either a plain list or a paginated object with a "results" key
resp.raise_for_status() releases = data if isinstance(data, list) else data.get("results", [])
data = resp.json() return [r["id"] for r in releases]
for r in data.get("results", []):
ids.append(r["id"])
paging = data.get("paging", {})
if page >= paging.get("totalPages", 1):
break
page += 1
return ids
def download_release_zip(release_id: str) -> zipfile.ZipFile: def download_release_zip(release_id: str) -> zipfile.ZipFile: