From 4c1f227f800136f0dc4aa79f477d44914a25b4ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bruno=20Adel=C3=A9?= Date: Sat, 10 Oct 2026 23:18:27 +0200 Subject: [PATCH 1/2] feat: vigilo refactoring --- importer/release_cloudflare.sh | 159 ++---------------- importer/vigilo/README.md | 67 ++++++++ importer/vigilo/__init__.py | 1 + importer/vigilo/_data2duckdb.sql | 6 +- importer/wdalib.py | 173 +++++++++++++++++++- justfile | 1 + requirements.txt | 3 + tests/test_wdalib_r2.py | 271 +++++++++++++++++++++++++++++++ 8 files changed, 528 insertions(+), 153 deletions(-) create mode 100644 importer/vigilo/README.md create mode 100644 tests/test_wdalib_r2.py diff --git a/importer/release_cloudflare.sh b/importer/release_cloudflare.sh index ce97158..c292b08 100755 --- a/importer/release_cloudflare.sh +++ b/importer/release_cloudflare.sh @@ -1,162 +1,27 @@ #!/usr/bin/env bash -# Upload parquets to Cloudflare R2 (S3-compatible) with parallel uploads. +# Upload parquets to Cloudflare R2 via wdalib.uploadToR2(). # Usage: DATAS_LIST="sirene,geonames" just release-cloudflare # # Required env vars: # CF_ACCOUNT_ID — Cloudflare account ID # CF_R2_ACCESS_KEY_ID — R2 API token Access Key ID # CF_R2_SECRET_ACCESS_KEY — R2 API token Secret Access Key -# -# Optional: -# R2_PARALLEL — number of parallel uploads (default: 8) -# -# Naming convention in R2: {dataset}/{subdir}__{path separators → __}{= → .} -# Example: sirene/observable__ape.46.38Z__data_0.parquet set -e -R2_BUCKET="world-datas-analysis" -R2_PARALLEL="${R2_PARALLEL:-8}" - -if [ -z "$CF_ACCOUNT_ID" ] || [ -z "$CF_R2_ACCESS_KEY_ID" ] || [ -z "$CF_R2_SECRET_ACCESS_KEY" ]; then - echo "[release-r2] ERROR: missing credentials — set CF_ACCOUNT_ID, CF_R2_ACCESS_KEY_ID, CF_R2_SECRET_ACCESS_KEY" - exit 1 -fi - -if ! command -v aws &>/dev/null; then - echo "[release-r2] ERROR: aws CLI not found (pip install awscli)" - exit 1 -fi - datasets=${DATAS_LIST:-""} -datasets=${datasets//,/ } - if [ -z "$datasets" ]; then - echo "[release-r2] DATAS_LIST is empty, nothing to do" + echo "[uploadToR2] DATAS_LIST is empty, nothing to do" exit 0 fi -R2_ENDPOINT="https://${CF_ACCOUNT_ID}.r2.cloudflarestorage.com" - -# Export credentials + config once — inherited by all xargs subshells -export AWS_ACCESS_KEY_ID="$CF_R2_ACCESS_KEY_ID" -export AWS_SECRET_ACCESS_KEY="$CF_R2_SECRET_ACCESS_KEY" -export R2_ENDPOINT R2_BUCKET - -for dataset in $datasets; do - declare -A local_keys # r2_key → local_file_size - declare -A r2_sizes # r2_key → r2_object_size - - # Quick local-hash check — skip entirely if parquets unchanged since last sync - sync_cache="dataset/${dataset}/.r2-sync-hash" - current_hash=$(find "dataset/${dataset}" \( -name "*.parquet" \) -printf '%s %T@\n' 2>/dev/null | sort | sha256sum | cut -d' ' -f1) - if [ -f "$sync_cache" ] && [ "$(cat "$sync_cache" 2>/dev/null)" = "$current_hash" ]; then - echo "[release-r2] ${dataset}: parquets unchanged since last sync, skipping" - continue - fi - - # Fetch current R2 state once — used for both skip-upload and stale-delete - echo "[release-r2] ${dataset}: listing R2 objects..." - while IFS= read -r line; do - [ -z "$line" ] && continue - r2_size=$(echo "$line" | awk '{print $3}') - r2_key=$(echo "$line" | awk '{print $4}') - r2_sizes["$r2_key"]=$r2_size - done < <( - aws s3 ls "s3://${R2_BUCKET}/${dataset}/" \ - --endpoint-url "$R2_ENDPOINT" \ - --recursive - ) - - task_file=$(mktemp) - skipped=0 - uploaded=0 - - for subdir in raw observable; do - src_dir="dataset/${dataset}/${subdir}" - [ -d "$src_dir" ] || continue - - while IFS= read -r f; do - rel=$(realpath --relative-to="dataset/${dataset}" "$f") - flat=$(echo "$rel" | sed 's|/|__|g') - r2_key="${dataset}/$(echo "$flat" | sed 's|=|\.|g')" - local_size=$(wc -c <"$f" | tr -d ' ') - local_keys["$r2_key"]=$local_size - - # Skip upload if R2 already has the same size - if [ "${r2_sizes[$r2_key]+_}" ] && [ "${r2_sizes[$r2_key]}" -eq "$local_size" ]; then - skipped=$((skipped + 1)) - continue - fi - - printf '%s\0%s\0' "$f" "$r2_key" >>"$task_file" - uploaded=$((uploaded + 1)) - done < <(find "$src_dir" -name "*.parquet" | sort) - done - - if [ "$skipped" -gt 0 ]; then - echo "[release-r2] ${dataset}: $skipped file(s) unchanged, skipped" - fi - - if [ "$uploaded" -eq 0 ] && [ "$skipped" -eq 0 ]; then - echo "[release-r2] ${dataset}: no raw/ or observable/ directory found, skipping" - elif [ "$uploaded" -gt 0 ]; then - echo "[release-r2] ${dataset}: uploading $uploaded new/changed file(s) (parallel=${R2_PARALLEL})..." - progress_file=$(mktemp) - export PROGRESS_FILE="$progress_file" - - ( - while true; do - done=$(wc -l <"$progress_file" 2>/dev/null | tr -d ' ') - printf "\r[release-r2] %d / %d" "$done" "$uploaded" - [ "$done" -ge "$uploaded" ] && break - sleep 0.2 - done - ) & - progress_pid=$! - - xargs -0 -P "$R2_PARALLEL" -n 2 bash -c \ - 'aws s3 cp "$1" "s3://${R2_BUCKET}/$2" --endpoint-url "$R2_ENDPOINT" --quiet && echo >> "$PROGRESS_FILE"' _ \ - <"$task_file" - - wait "$progress_pid" 2>/dev/null - printf "\r[release-r2] ${dataset}: done (%d files uploaded)\n" "$uploaded" - rm -f "$progress_file" - fi - rm -f "$task_file" - - # Delete R2 keys that no longer exist locally (batch DeleteObjects, 1000/req) - echo "[release-r2] ${dataset}: checking for stale R2 files..." - stale_keys=() - for r2_key in "${!r2_sizes[@]}"; do - if [ -z "${local_keys[$r2_key]+_}" ]; then - stale_keys+=("$r2_key") - fi - done - - deleted=${#stale_keys[@]} - if [ "$deleted" -gt 0 ]; then - echo "[release-r2] ${dataset}: deleting $deleted stale file(s) in batches of 1000..." - i=0 - while [ $i -lt $deleted ]; do - batch=("${stale_keys[@]:$i:1000}") - json=$(printf '{"Key":"%s"},' "${batch[@]}") - json="{\"Objects\":[${json%,}],\"Quiet\":true}" - aws s3api delete-objects \ - --bucket "$R2_BUCKET" \ - --endpoint-url "$R2_ENDPOINT" \ - --delete "$json" \ - --output text > /dev/null - i=$((i + 1000)) - echo "[release-r2] deleted $((i < deleted ? i : deleted)) / $deleted" - done - echo "[release-r2] ${dataset}: $deleted stale file(s) deleted" - else - echo "[release-r2] ${dataset}: no stale files" - fi - - # Save hash so next run skips if nothing changed - echo "$current_hash" > "$sync_cache" - - unset local_keys r2_sizes -done +PYTHON="${PYTHON:-.venv/bin/python3}" +"$PYTHON" - "$datasets" <<'PYEOF' +import sys +sys.path.insert(0, 'importer') +import wdalib +for d in sys.argv[1].split(','): + d = d.strip() + if d: + wdalib.uploadToR2(d) +PYEOF diff --git a/importer/vigilo/README.md b/importer/vigilo/README.md new file mode 100644 index 0000000..84a3ee5 --- /dev/null +++ b/importer/vigilo/README.md @@ -0,0 +1,67 @@ +# Vigilo dataset + +## Overview + +Vigilo is a citizen reporting platform for cycling infrastructure issues in +French cities. Each city runs its own Vigilo instance (called a _scope_). + +## Workflow + +``` +just download → just update → just release-cloudflare +``` + +### 1. `just download` + +Fetches live data from the Vigilo API (runs inside Docker): + +- `downloaded/vigilo/categories.json` — issue categories +- `downloaded/vigilo/scopes.json` — active city scopes +- `downloaded/vigilo/observations_.json` — observations per scope + +Only runs if the downloaded folder is older than 7 days. + +### 2. `just update` + +Merges new API data with historical data from R2 (runs inside Docker): + +1. **`downloadFromR2`** — downloads the latest `scopes.parquet` and + `observations.parquet` from R2 into `downloaded/vigilo/from_r2/raw/` using + the manifest. This ensures the merge always starts from the most recent + published state. +2. **`_data2duckdb.sql`** — merges new observations with historical data: + - Active scopes: `first_seen_at` preserved from R2 history + - Inactive scopes (disappeared from API): kept with `is_active = false` + - Observations from inactive scopes: preserved from R2 history +3. Outputs to `dataset/vigilo/raw/` + +### 3. `just release-cloudflare` + +Uploads the generated parquets to Cloudflare R2 (runs **locally**, needs R2 +credentials). Also generates `vigilo/manifest-vigilo.json` on R2, which is used +by the next `just update` to download the historical data. + +Required env vars: + +Stored on `.env` file + +``` +CF_ACCOUNT_ID +CF_R2_ACCESS_KEY_ID +CF_R2_SECRET_ACCESS_KEY +``` + +## R2 naming convention + +Local path → R2 key: `/` replaced by `__`, `=` replaced by `.` + +``` +dataset/vigilo/raw/scopes.parquet → vigilo/raw__scopes.parquet +``` + +## First run (no prior R2 data) + +If no manifest exists on R2 yet, `downloadFromR2` skips gracefully and the SQL +runs without historical data (all scopes treated as new, +`first_seen_at = today`). Run `just release-cloudflare` after the first +`just update` to publish data and create the manifest. diff --git a/importer/vigilo/__init__.py b/importer/vigilo/__init__.py index 82c4a2c..85a0a29 100644 --- a/importer/vigilo/__init__.py +++ b/importer/vigilo/__init__.py @@ -131,4 +131,5 @@ def download(): def update(): wdalib.show_title(f"Update {provider}") + wdalib.downloadFromR2(provider, subdir="raw") wdalib.data2duckdb(provider) diff --git a/importer/vigilo/_data2duckdb.sql b/importer/vigilo/_data2duckdb.sql index e39cbff..7fa7417 100644 --- a/importer/vigilo/_data2duckdb.sql +++ b/importer/vigilo/_data2duckdb.sql @@ -60,7 +60,7 @@ INSERT INTO vigilo_scopes BY NAME true AS is_active, COALESCE( (SELECT first_seen_at - FROM read_parquet('./dataset/vigilo/raw/scopes.parquet') + FROM read_parquet('./downloaded/vigilo/from_r2/raw/scopes.parquet') WHERE id = n.id LIMIT 1), CURRENT_DATE ) AS first_seen_at, @@ -72,7 +72,7 @@ INSERT INTO vigilo_scopes BY NAME -- On force is_active = false quel que soit leur statut précédent (gère la première disparition) INSERT INTO vigilo_scopes BY NAME SELECT * REPLACE (false AS is_active) - FROM read_parquet('./dataset/vigilo/raw/scopes.parquet') + FROM read_parquet('./downloaded/vigilo/from_r2/raw/scopes.parquet') WHERE id NOT IN (SELECT id FROM vigilo_scopes_new); ------------------------------------------------------------------------------- @@ -109,7 +109,7 @@ INSERT INTO vigilo_observations SELECT scopeid, token, ts, latitude, longitude, address, "comment", explanation, catid, approved, cityname, geonames_districtid, geonames_district, geonames_cityid, geonames_city - FROM read_parquet('./dataset/vigilo/raw/observations.parquet') + FROM read_parquet('./downloaded/vigilo/from_r2/raw/observations.parquet') WHERE scopeid NOT IN (SELECT DISTINCT scopeid FROM vigilo_observations); ------------------------------------------------------------------------------- diff --git a/importer/wdalib.py b/importer/wdalib.py index 4e01dc2..17b7a8b 100644 --- a/importer/wdalib.py +++ b/importer/wdalib.py @@ -3,14 +3,12 @@ import datetime import glob import hashlib +import json import os import shutil import subprocess import zipfile -import git -import requests -from tqdm import tqdm def init_download(provider): @@ -117,6 +115,8 @@ def isFolderOutdated(folder, hours): # Clone or pull repository ############################################################################### def pull(provider, repo_url): + import git + clone_dir = f"./downloaded/{provider}" print(f"Sync {provider} repository ...") @@ -168,6 +168,9 @@ def data2duckdb(provider, tables=None): def download(url, desc="", save_path=""): + import requests + from tqdm import tqdm + # Init stream downloader response = requests.get(url, stream=True) response.raise_for_status() @@ -227,6 +230,8 @@ def _get_github_repo(): def download_last_dataset(provider): + import requests + tag = f"dataset-{provider}" dest = f"./downloaded/{provider}/last_release" @@ -264,3 +269,165 @@ def unzipFile(filename, destination): with zipfile.ZipFile(filename, "r") as zip_ref: zip_ref.extractall(destination) os.remove(filename) + + +############################################################################### +# Cloudflare R2 upload / download +# +# Naming convention in R2: {dataset}/{subdir}__{path separators → __}{= → .} +# Example: dataset/vigilo, file raw/scopes.parquet → vigilo/raw__scopes.parquet +# +# uploadToR2 — runs locally (hors Docker), needs R2 credentials +# downloadFromR2 — runs inside Docker, uses the public R2 URL +############################################################################### + +_R2_BUCKET = "world-datas-analysis" + + +def uploadToR2(dataset: str) -> None: + import boto3 + + account_id = os.environ.get("CF_ACCOUNT_ID") + access_key = os.environ.get("CF_R2_ACCESS_KEY_ID") + secret_key = os.environ.get("CF_R2_SECRET_ACCESS_KEY") + + if not account_id or not access_key or not secret_key: + raise RuntimeError( + "[uploadToR2] Missing credentials: " + "CF_ACCOUNT_ID, CF_R2_ACCESS_KEY_ID, CF_R2_SECRET_ACCESS_KEY" + ) + + files = [] + for subdir in ["raw", "observable"]: + src_dir = f"dataset/{dataset}/{subdir}" + if not os.path.isdir(src_dir): + continue + for root, _, filenames in os.walk(src_dir): + for fn in sorted(filenames): + if fn.endswith(".parquet"): + files.append(os.path.join(root, fn)) + files.sort() + + if not files: + print(f"[uploadToR2] {dataset}: no parquet files found in raw/ or observable/") + return + + endpoint_url = f"https://{account_id}.r2.cloudflarestorage.com" + client = boto3.client( + "s3", + endpoint_url=endpoint_url, + aws_access_key_id=access_key, + aws_secret_access_key=secret_key, + ) + + sync_cache = f"dataset/{dataset}/.r2-sync-hash" + hash_input = "\n".join( + f"{os.path.getsize(f):.0f} {os.path.getmtime(f):.6f}" for f in files + ) + current_hash = hashlib.sha256(hash_input.encode()).hexdigest() + if os.path.exists(sync_cache): + with open(sync_cache) as f: + if f.read().strip() == current_hash: + print(f"[uploadToR2] {dataset}: parquets unchanged since last sync, skipping") + return + + r2_sizes = {} + paginator = client.get_paginator("list_objects_v2") + for page in paginator.paginate(Bucket=_R2_BUCKET, Prefix=f"{dataset}/"): + for obj in page.get("Contents", []): + r2_sizes[obj["Key"]] = obj["Size"] + + manifest = {} + local_keys = set() + uploaded = 0 + skipped = 0 + + for f in files: + rel = os.path.relpath(f, f"dataset/{dataset}") + flat = rel.replace("/", "__").replace("=", ".") + r2_key = f"{dataset}/{flat}" + local_keys.add(r2_key) + manifest[flat] = rel + + local_size = os.path.getsize(f) + if r2_key in r2_sizes and r2_sizes[r2_key] == local_size: + skipped += 1 + continue + + with open(f, "rb") as fp: + client.put_object(Bucket=_R2_BUCKET, Key=r2_key, Body=fp.read()) + uploaded += 1 + + manifest_key = f"{dataset}/manifest-{dataset}.json" + client.put_object( + Bucket=_R2_BUCKET, + Key=manifest_key, + Body=json.dumps(manifest).encode(), + ContentType="application/json", + ) + local_keys.add(manifest_key) + print(f"[uploadToR2] {dataset}: manifest uploaded ({len(manifest)} file(s) listed)") + + if skipped: + print(f"[uploadToR2] {dataset}: {skipped} file(s) unchanged, skipped") + if uploaded: + print(f"[uploadToR2] {dataset}: {uploaded} file(s) uploaded") + + stale = [k for k in r2_sizes if k not in local_keys] + if stale: + for i in range(0, len(stale), 1000): + batch = stale[i : i + 1000] + client.delete_objects( + Bucket=_R2_BUCKET, + Delete={"Objects": [{"Key": k} for k in batch], "Quiet": True}, + ) + print(f"[uploadToR2] {dataset}: {len(stale)} stale file(s) deleted") + else: + print(f"[uploadToR2] {dataset}: no stale files") + + with open(sync_cache, "w") as f: + f.write(current_hash) + + +def downloadFromR2(dataset: str, subdir: str = "raw") -> None: + import requests + + base_url = os.environ.get("WDA_PUBLIC_DATASET_URL", "").rstrip("/") + if not base_url: + raise RuntimeError("[downloadFromR2] WDA_PUBLIC_DATASET_URL is not set") + + manifest_url = f"{base_url}/{dataset}/manifest-{dataset}.json" + resp = requests.get(manifest_url, timeout=30) + if resp.status_code == 404: + print( + f"[downloadFromR2] {dataset}: no manifest found at {manifest_url}, " + "skipping (run 'just release' first)" + ) + return + resp.raise_for_status() + manifest = resp.json() + + prefix = f"{subdir}/" + dest_base = f"./downloaded/{dataset}/from_r2" + downloaded_count = 0 + + for flat_name, rel_path in manifest.items(): + if not rel_path.startswith(prefix): + continue + + file_url = f"{base_url}/{dataset}/{flat_name}" + dest_path = os.path.join(dest_base, rel_path) + os.makedirs(os.path.dirname(dest_path), exist_ok=True) + + file_resp = requests.get(file_url, timeout=60, stream=True) + file_resp.raise_for_status() + with open(dest_path, "wb") as f: + for chunk in file_resp.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + downloaded_count += 1 + + print( + f"[downloadFromR2] {dataset}/{subdir}: " + f"{downloaded_count} file(s) downloaded to {dest_base}" + ) diff --git a/justfile b/justfile index c892107..09772ee 100644 --- a/justfile +++ b/justfile @@ -270,6 +270,7 @@ release-delete: # Build the production site and serve it locally on port 8080 (mirrors CI) [group('observable')] @observable-prod: + echo "You can access the production site at http://localhost:8080" DATAS_LIST="${DATAS_LIST:-vigilo,nafrev2,sirene}" just observable-pages-build docker run --rm -p 8080:80 \ -v "$(pwd)/observable/dist:/usr/share/nginx/html:ro" \ diff --git a/requirements.txt b/requirements.txt index 9ab9cea..49bb86d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,6 +11,9 @@ tqdm==4.66.4 # Fix json json_repair==0.25.2 +# R2 / S3-compatible storage +boto3==1.35.36 + # Tests pytest==8.3.2 pytest-rich==0.1.1 diff --git a/tests/test_wdalib_r2.py b/tests/test_wdalib_r2.py new file mode 100644 index 0000000..479a2bd --- /dev/null +++ b/tests/test_wdalib_r2.py @@ -0,0 +1,271 @@ +"""Unit tests for wdalib.uploadToR2 and wdalib.downloadFromR2.""" + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, call, patch + +sys.path.insert(0, str(Path(__file__).parents[1] / "importer")) + +import wdalib + +_R2_CREDS = { + "CF_ACCOUNT_ID": "test-account", + "CF_R2_ACCESS_KEY_ID": "test-key", + "CF_R2_SECRET_ACCESS_KEY": "test-secret", +} + + +def _make_boto_client(r2_objects=None): + """Return a mock boto3 client with an empty or given R2 object listing.""" + client = MagicMock() + paginator = MagicMock() + client.get_paginator.return_value = paginator + contents = [{"Key": k, "Size": s} for k, s in (r2_objects or {}).items()] + paginator.paginate.return_value = [{"Contents": contents}] + return client + + +# --------------------------------------------------------------------------- +# uploadToR2 +# --------------------------------------------------------------------------- + + +class UploadToR2Tests(unittest.TestCase): + def _run_upload(self, dataset, mock_client, extra_env=None): + env = {**_R2_CREDS, **(extra_env or {})} + with patch("boto3.client", return_value=mock_client), patch.dict(os.environ, env): + wdalib.uploadToR2(dataset) + + def test_flattening_slash_to_double_underscore(self): + """raw/observations.parquet → {dataset}/raw__observations.parquet""" + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + parquet = Path("dataset/vigilo/raw/observations.parquet") + parquet.parent.mkdir(parents=True) + parquet.write_bytes(b"PAR1fake") + + client = _make_boto_client() + self._run_upload("vigilo", client) + + put_calls = [c for c in client.put_object.call_args_list] + keys = [c.kwargs["Key"] for c in put_calls] + self.assertIn("vigilo/raw__observations.parquet", keys) + finally: + os.chdir(orig) + + def test_equals_replaced_by_dot_in_hive_partition(self): + """raw/section_id=A/data.parquet → vigilo/raw__section_id.A__data.parquet""" + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + parquet = Path("dataset/vigilo/raw/section_id=A/data.parquet") + parquet.parent.mkdir(parents=True) + parquet.write_bytes(b"PAR1fake") + + client = _make_boto_client() + self._run_upload("vigilo", client) + + keys = [c.kwargs["Key"] for c in client.put_object.call_args_list] + self.assertIn("vigilo/raw__section_id.A__data.parquet", keys) + finally: + os.chdir(orig) + + def test_manifest_uploaded_after_files(self): + """manifest-{dataset}.json is uploaded and contains correct flat→rel mapping.""" + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + parquet = Path("dataset/vigilo/raw/scopes.parquet") + parquet.parent.mkdir(parents=True) + parquet.write_bytes(b"PAR1fake") + + client = _make_boto_client() + self._run_upload("vigilo", client) + + manifest_call = next( + c for c in client.put_object.call_args_list + if c.kwargs["Key"] == "vigilo/manifest-vigilo.json" + ) + manifest = json.loads(manifest_call.kwargs["Body"]) + self.assertEqual(manifest["raw__scopes.parquet"], "raw/scopes.parquet") + finally: + os.chdir(orig) + + def test_skip_file_when_r2_size_matches(self): + """A file whose size already matches R2 is not re-uploaded.""" + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + parquet = Path("dataset/vigilo/raw/scopes.parquet") + parquet.parent.mkdir(parents=True) + parquet.write_bytes(b"PAR1fake") + + r2_objects = {"vigilo/raw__scopes.parquet": len(b"PAR1fake")} + client = _make_boto_client(r2_objects) + self._run_upload("vigilo", client) + + keys = [c.kwargs["Key"] for c in client.put_object.call_args_list] + self.assertNotIn("vigilo/raw__scopes.parquet", keys) + self.assertIn("vigilo/manifest-vigilo.json", keys) + finally: + os.chdir(orig) + + def test_stale_r2_keys_deleted(self): + """R2 keys that no longer exist locally trigger delete_objects.""" + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + parquet = Path("dataset/vigilo/raw/scopes.parquet") + parquet.parent.mkdir(parents=True) + parquet.write_bytes(b"PAR1fake") + + r2_objects = { + "vigilo/raw__scopes.parquet": 0, + "vigilo/raw__old_file.parquet": 9999, + } + client = _make_boto_client(r2_objects) + self._run_upload("vigilo", client) + + client.delete_objects.assert_called_once() + deleted = client.delete_objects.call_args.kwargs["Delete"]["Objects"] + deleted_keys = [d["Key"] for d in deleted] + self.assertIn("vigilo/raw__old_file.parquet", deleted_keys) + self.assertNotIn("vigilo/raw__scopes.parquet", deleted_keys) + finally: + os.chdir(orig) + + def test_missing_credentials_raises(self): + """Raises RuntimeError when R2 credentials are absent.""" + env = {k: v for k, v in os.environ.items() + if k not in ("CF_ACCOUNT_ID", "CF_R2_ACCESS_KEY_ID", "CF_R2_SECRET_ACCESS_KEY")} + with patch.dict(os.environ, env, clear=True): + with self.assertRaises(RuntimeError, msg="should raise without credentials"): + wdalib.uploadToR2("vigilo") + + def test_no_parquet_files_returns_early(self): + """Returns without calling boto3 if no parquet files exist.""" + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + Path("dataset/vigilo/raw").mkdir(parents=True) + + with patch("boto3.client") as mock_boto3_client, \ + patch.dict(os.environ, _R2_CREDS): + wdalib.uploadToR2("vigilo") + mock_boto3_client.assert_not_called() + finally: + os.chdir(orig) + + +# --------------------------------------------------------------------------- +# downloadFromR2 +# --------------------------------------------------------------------------- + + +class DownloadFromR2Tests(unittest.TestCase): + _ENV = {"WDA_PUBLIC_DATASET_URL": "https://pub.example.r2.dev"} + + def _manifest_response(self, manifest): + r = MagicMock() + r.status_code = 200 + r.raise_for_status = MagicMock() + r.json.return_value = manifest + return r + + def _file_response(self, content=b"PAR1fake"): + r = MagicMock() + r.status_code = 200 + r.raise_for_status = MagicMock() + r.iter_content.return_value = [content] + return r + + def test_files_written_to_from_r2_subdir(self): + """Parquet files are written to downloaded/{dataset}/from_r2/raw/.""" + manifest = { + "raw__scopes.parquet": "raw/scopes.parquet", + "raw__observations.parquet": "raw/observations.parquet", + } + responses = [ + self._manifest_response(manifest), + self._file_response(), + self._file_response(), + ] + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + with patch("requests.get", side_effect=responses), \ + patch.dict(os.environ, self._ENV): + wdalib.downloadFromR2("vigilo", subdir="raw") + + self.assertTrue(Path("downloaded/vigilo/from_r2/raw/scopes.parquet").exists()) + self.assertTrue(Path("downloaded/vigilo/from_r2/raw/observations.parquet").exists()) + finally: + os.chdir(orig) + + def test_manifest_url_constructed_correctly(self): + """Manifest URL uses WDA_PUBLIC_DATASET_URL + dataset prefix.""" + r404 = MagicMock() + r404.status_code = 404 + r404.raise_for_status = MagicMock() + + with patch("requests.get", return_value=r404) as mock_get, \ + patch.dict(os.environ, self._ENV): + wdalib.downloadFromR2("vigilo") + + mock_get.assert_called_once_with( + "https://pub.example.r2.dev/vigilo/manifest-vigilo.json", + timeout=30, + ) + + def test_observable_files_excluded_when_subdir_raw(self): + """observable/ files are not downloaded when subdir='raw'.""" + manifest = { + "raw__scopes.parquet": "raw/scopes.parquet", + "observable__map.parquet": "observable/map.parquet", + } + responses = [self._manifest_response(manifest), self._file_response()] + with tempfile.TemporaryDirectory() as tmpdir: + orig = os.getcwd() + os.chdir(tmpdir) + try: + with patch("requests.get", side_effect=responses), \ + patch.dict(os.environ, self._ENV): + wdalib.downloadFromR2("vigilo", subdir="raw") + + self.assertTrue(Path("downloaded/vigilo/from_r2/raw/scopes.parquet").exists()) + self.assertFalse(Path("downloaded/vigilo/from_r2/observable/map.parquet").exists()) + finally: + os.chdir(orig) + + def test_missing_manifest_404_returns_gracefully(self): + """A 404 on the manifest does not raise, just prints a warning.""" + r = MagicMock() + r.status_code = 404 + r.raise_for_status = MagicMock() + + with patch("requests.get", return_value=r), \ + patch.dict(os.environ, self._ENV): + wdalib.downloadFromR2("vigilo") # must not raise + + def test_missing_env_var_raises(self): + """Raises RuntimeError when WDA_PUBLIC_DATASET_URL is not set.""" + env = {k: v for k, v in os.environ.items() if k != "WDA_PUBLIC_DATASET_URL"} + with patch.dict(os.environ, env, clear=True): + with self.assertRaises(RuntimeError): + wdalib.downloadFromR2("vigilo") + + +if __name__ == "__main__": + unittest.main() From c3050d50756171618c0f2606fec8869de0af621b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bruno=20Adel=C3=A9?= Date: Sat, 10 Oct 2026 23:29:01 +0200 Subject: [PATCH 2/2] feat: update justfile version --- .github/workflows/docker-image.yml | 2 +- .github/workflows/release-datasets.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index cee7346..9ff1e23 100644 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -6,7 +6,7 @@ on: - "Dockerfile" - "requirements.txt" env: - JUST_VERSION: "1.23.0" + JUST_VERSION: "1.51.0" jobs: push_to_registry: diff --git a/.github/workflows/release-datasets.yml b/.github/workflows/release-datasets.yml index 01d9b29..b963c71 100644 --- a/.github/workflows/release-datasets.yml +++ b/.github/workflows/release-datasets.yml @@ -12,7 +12,7 @@ permissions: contents: write env: - JUST_VERSION: "1.23.0" + JUST_VERSION: "1.51.0" jobs: release: