From 71b24cfd8318dd0551c94b9f60212d390bc8485c Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sat, 29 Aug 2026 19:30:19 -0400 Subject: [PATCH 1/3] Add NZ and UK source storage conventions (#175) Route new raw and derived artifacts through country-aware publisher prefixes, refuse stale manifest keys, and record NZ period/geography rulings. Preserve legacy US paths and leave UK object copying and manifest migration to a separate credentialed change. --- chronicle/artifacts.py | 176 ++++++++++++++++++++++++++++-- chronicle/core.py | 8 +- chronicle/harness.py | 27 +++-- docs/pe-nz-source-checklist.md | 67 ++++++++++++ docs/storage-architecture.md | 30 +++-- tests/test_chronicle_artifacts.py | 121 ++++++++++++++++++++ 6 files changed, 396 insertions(+), 33 deletions(-) create mode 100644 docs/pe-nz-source-checklist.md diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py index 139663c5..c2b944fa 100644 --- a/chronicle/artifacts.py +++ b/chronicle/artifacts.py @@ -24,6 +24,31 @@ DEFAULT_R2_PREFIX = "raw" DEFAULT_R2_DERIVED_PREFIX = "derived" +# New UK and New Zealand uploads are namespaced by country. US objects predate +# the country segment and deliberately keep their legacy ``raw/{source_id}`` +# and ``derived/{source_id}`` shapes. Publisher directories are the stable +# routing input because they are shared by packages/ and db/data/. +R2_COUNTRY_PUBLISHERS = { + "nz": frozenset({"ird", "mbie", "msd", "stats_nz"}), + "uk": frozenset( + { + "dft", + "dwp", + "hmrc", + "isc", + "mhclg", + "nisra", + "nrs", + "obr", + "ons", + "scotgov", + "slc", + "voa", + "welshgov", + } + ), +} + @dataclass(frozen=True) class ArtifactStorageLocation: @@ -366,7 +391,7 @@ def fetch_source_artifact( filename: str | None = None, upload_r2: bool = False, r2_bucket: str = DEFAULT_R2_RAW_BUCKET, - r2_prefix: str = DEFAULT_R2_PREFIX, + r2_prefix: str | None = None, wrangler_command: str = "npx wrangler", ) -> ArtifactFetchReport: """Fetch/register a source artifact and optionally upload it to R2.""" @@ -385,6 +410,12 @@ def fetch_source_artifact( size_bytes = len(content) manifest_path = output / "manifest.yaml" + resolved_r2_prefix = resolve_r2_prefix( + prefix=r2_prefix, + default_prefix=DEFAULT_R2_PREFIX, + source_id=source_id, + package_path=output, + ) r2_location = ArtifactStorageLocation( provider="r2", bucket=r2_bucket, @@ -394,7 +425,7 @@ def fetch_source_artifact( year=year, sha256=sha256, filename=artifact_filename, - prefix=r2_prefix, + prefix=resolved_r2_prefix, ), ) r2_upload = None @@ -449,7 +480,7 @@ def publish_derived_artifacts( year: int, build_id: str | None = None, r2_bucket: str = DEFAULT_R2_DERIVED_BUCKET, - r2_prefix: str = DEFAULT_R2_DERIVED_PREFIX, + r2_prefix: str | None = None, wrangler_command: str = "npx wrangler", build_artifacts_output: str | Path | None = None, ) -> DerivedArtifactPublishReport: @@ -497,6 +528,11 @@ def publish_derived_artifacts( errors=("missing_build_id",), ) + resolved_r2_prefix = resolve_r2_prefix( + prefix=r2_prefix, + default_prefix=DEFAULT_R2_DERIVED_PREFIX, + source_id=source_id, + ) entries: list[DerivedArtifactUploadEntry] = [] errors: list[str] = [] artifact_paths = sorted(path for path in input_path.rglob("*") if path.is_file()) @@ -515,7 +551,7 @@ def publish_derived_artifacts( year=year, build_id=resolved_build_id, artifact_name=relative_path, - prefix=r2_prefix, + prefix=resolved_r2_prefix, ), ) upload = _upload_r2_object( @@ -560,7 +596,7 @@ def publish_source_artifacts( source_id: str | None = None, package_id: str | None = None, r2_bucket: str = DEFAULT_R2_RAW_BUCKET, - r2_prefix: str = DEFAULT_R2_PREFIX, + r2_prefix: str | None = None, wrangler_command: str = "npx wrangler", ) -> RawArtifactPublishReport: """Upload manifest-declared raw source artifacts and record R2 locations.""" @@ -594,6 +630,17 @@ def publish_source_artifacts( errors.append(f"Manifest files must be a mapping: {manifest_path}") continue + try: + resolved_r2_prefix = resolve_r2_prefix( + prefix=r2_prefix, + default_prefix=DEFAULT_R2_PREFIX, + source_id=str(manifest_source_id), + package_path=manifest_path, + ) + except ValueError as exc: + errors.append(f"Could not resolve R2 prefix for {manifest_path}: {exc}") + continue + updated = False for year, spec in files.items(): entry, updated_spec = _publish_raw_manifest_entry( @@ -603,7 +650,7 @@ def publish_source_artifacts( year, spec, r2_bucket=r2_bucket, - r2_prefix=r2_prefix, + r2_prefix=resolved_r2_prefix, wrangler_command=wrangler_command, ) entries.append(entry) @@ -764,6 +811,81 @@ def bootstrap_r2_buckets( ) +def infer_r2_country( + *, + source_id: str | None = None, + package_path: str | Path | None = None, +) -> str | None: + """Infer an R2 country segment from a package publisher directory. + + ``package_path`` is authoritative when supplied. ``source_id`` is a + fallback for low-level callers that do not have a package path, including + derived-artifact publication. + """ + path_countries: set[str] = set() + if package_path is not None: + path_parts = {part.lower() for part in Path(package_path).parts} + path_countries = { + country + for country, publishers in R2_COUNTRY_PUBLISHERS.items() + if path_parts.intersection(publishers) + } + if len(path_countries) > 1: + raise ValueError( + "package path contains publishers from multiple countries: " + f"{package_path}" + ) + + source_countries: set[str] = set() + normalized_source_id = (source_id or "").lower() + if normalized_source_id: + source_countries = { + country + for country, publishers in R2_COUNTRY_PUBLISHERS.items() + if any( + normalized_source_id == publisher + or normalized_source_id.startswith(f"{publisher}_") + or normalized_source_id.startswith(f"{publisher}-") + for publisher in publishers + ) + } + if len(source_countries) > 1: + raise ValueError(f"source_id maps to multiple R2 countries: {source_id}") + + if path_countries and source_countries and path_countries != source_countries: + raise ValueError( + "package publisher directory and source_id map to different " + f"countries: path={sorted(path_countries)}, " + f"source_id={sorted(source_countries)}" + ) + countries = path_countries or source_countries + return next(iter(countries)) if countries else None + + +def resolve_r2_prefix( + *, + prefix: str | None, + default_prefix: str, + source_id: str | None = None, + package_path: str | Path | None = None, +) -> str: + """Return the country-aware R2 prefix for one source package.""" + resolved = _clean_key_part(prefix or default_prefix) + country = infer_r2_country(source_id=source_id, package_path=package_path) + if country is None: + return resolved + + suffix = resolved.rsplit("/", maxsplit=1)[-1] + if suffix in R2_COUNTRY_PUBLISHERS: + if suffix != country: + raise ValueError( + f"R2 prefix country {suffix!r} disagrees with publisher " + f"country {country!r}" + ) + return resolved + return posixpath.join(resolved, country) + + def build_r2_key( *, source_id: str, @@ -771,15 +893,20 @@ def build_r2_key( year: int | str, sha256: str, filename: str, - prefix: str = DEFAULT_R2_PREFIX, + prefix: str | None = None, ) -> str: """Build the canonical immutable R2 key for a raw source artifact. ``year`` is usually a calendar year but may be a label such as ``source_capture`` for non-year manifest file entries. """ + resolved_prefix = resolve_r2_prefix( + prefix=prefix, + default_prefix=DEFAULT_R2_PREFIX, + source_id=source_id, + ) return posixpath.join( - _clean_key_part(prefix), + resolved_prefix, _clean_key_part(source_id), _clean_key_part(package_id), str(year), @@ -795,11 +922,16 @@ def build_derived_r2_key( year: int, build_id: str, artifact_name: str, - prefix: str = DEFAULT_R2_DERIVED_PREFIX, + prefix: str | None = None, ) -> str: """Build the canonical R2 key for a derived build artifact.""" + resolved_prefix = resolve_r2_prefix( + prefix=prefix, + default_prefix=DEFAULT_R2_DERIVED_PREFIX, + source_id=source_id, + ) return posixpath.join( - _clean_key_part(prefix), + resolved_prefix, _clean_key_part(source_id), _clean_key_part(package_id), str(year), @@ -1002,6 +1134,30 @@ def _publish_raw_manifest_entry( prefix=r2_prefix, ), ) + storage = spec.get("storage") if isinstance(spec.get("storage"), dict) else {} + recorded_r2 = storage.get("r2") if isinstance(storage.get("r2"), dict) else {} + recorded_key = recorded_r2.get("key") + if recorded_key and recorded_key != location.key: + errors.append( + "recorded_r2_key_disagrees_with_country_prefix:" + f"recorded={recorded_key}:expected={location.key}" + ) + return ( + RawArtifactPublishEntry( + manifest_path=str(manifest_path), + source_id=source_id, + package_id=package_id, + year=str(year), + filename=filename, + local_path=str(artifact_path), + sha256=sha256_actual, + size_bytes=size_bytes, + r2_location=None, + upload=None, + errors=tuple(errors), + ), + None, + ) upload = _upload_r2_object( location, artifact_path, diff --git a/chronicle/core.py b/chronicle/core.py index c43f7574..16dcc3cf 100644 --- a/chronicle/core.py +++ b/chronicle/core.py @@ -21,8 +21,12 @@ # (packages/hmrc/vat_firm_targets_2024_25) and academic year AY 2024/25 is # academic_year 2024 (#131). Publishers that label a fiscal year with a single # year keep that label year (US federal FY2024 -> 2024, -# packages/cbo/individual_income_tax_receipts_2026_02). Caution: the EES -# helper _academic_year_end (chronicle/sources/rows.py) names value COLUMNS +# packages/cbo/individual_income_tax_receipts_2026_02). +# New Zealand labels its April-March income tax year by the ending year, so the +# publisher's "2024 tax year" is tax_year 2024 with PeriodCoverage 2023-04-01 +# through 2024-03-31 and the publisher label recorded in source_period_label. +# Caution: the EES helper _academic_year_end (chronicle/sources/rows.py) names +# value COLUMNS # by the academic year's END year ("2024/25" -> 2025); that is source-layout # naming only and must not leak into fact periods. ALLOWED_PERIOD_TYPES = { diff --git a/chronicle/harness.py b/chronicle/harness.py index e45f3604..1d96f093 100644 --- a/chronicle/harness.py +++ b/chronicle/harness.py @@ -336,7 +336,7 @@ def fetch_artifact_file( filename: str | None = None, upload_r2: bool = False, r2_bucket: str = "ledger-raw", - r2_prefix: str = "raw", + r2_prefix: str | None = None, wrangler_command: str = "npx wrangler", ) -> ArtifactFetchReport: """Fetch/register a raw source artifact and optionally upload it to R2.""" @@ -373,7 +373,7 @@ def publish_raw_artifact_files( source_id: str | None = None, package_id: str | None = None, r2_bucket: str = "ledger-raw", - r2_prefix: str = "raw", + r2_prefix: str | None = None, wrangler_command: str = "npx wrangler", ) -> RawArtifactPublishReport: """Publish manifest-declared raw source artifacts to R2.""" @@ -410,7 +410,7 @@ def publish_derived_artifact_files( year: int, build_id: str | None = None, r2_bucket: str = "ledger-derived", - r2_prefix: str = "derived", + r2_prefix: str | None = None, wrangler_command: str = "npx wrangler", build_artifacts_output: str | Path | None = None, ) -> DerivedArtifactPublishReport: @@ -893,8 +893,11 @@ def main(argv: list[str] | None = None) -> int: ) artifact_parser.add_argument( "--r2-prefix", - default="raw", - help="R2 key prefix for raw artifacts.", + help=( + "R2 key-prefix root for raw artifacts. Defaults to country-aware " + "raw/nz or raw/uk for known publisher directories, and raw for " + "legacy US packages." + ), ) artifact_parser.add_argument( "--wrangler-command", @@ -948,8 +951,11 @@ def main(argv: list[str] | None = None) -> int: ) raw_publish_parser.add_argument( "--r2-prefix", - default="raw", - help="R2 key prefix for raw artifacts.", + help=( + "R2 key-prefix root for raw artifacts. Defaults to country-aware " + "raw/nz or raw/uk for known publisher directories, and raw for " + "legacy US packages." + ), ) raw_publish_parser.add_argument( "--wrangler-command", @@ -1014,8 +1020,11 @@ def main(argv: list[str] | None = None) -> int: ) derived_publish_parser.add_argument( "--r2-prefix", - default="derived", - help="R2 key prefix for derived build artifacts.", + help=( + "R2 key-prefix root for derived artifacts. Defaults to " + "country-aware derived/nz or derived/uk for known publishers, " + "and derived for legacy US packages." + ), ) derived_publish_parser.add_argument( "--wrangler-command", diff --git a/docs/pe-nz-source-checklist.md b/docs/pe-nz-source-checklist.md new file mode 100644 index 00000000..ab98924f --- /dev/null +++ b/docs/pe-nz-source-checklist.md @@ -0,0 +1,67 @@ +# PolicyEngine New Zealand source checklist + +This checklist records Chronicle's New Zealand source-ingestion decisions. The +canonical wave-1 inventory and acceptance criteria live in +[issue #176](https://github.com/PolicyEngine/chronicle/issues/176); this file is +the repository execution ledger for that issue, not a separate target design. +Chronicle stores publisher-backed facts only. Microcosm owns target selection, +reconciliation, aging, and activation. + +## Period convention + +IRD labels New Zealand's April-March income tax year by its ending year. +Publisher label "2024 tax year" therefore becomes `tax_year: 2024`, with every +NZ tax-year record set also carrying: + +```yaml +period_coverage: + start_date: 2023-04-01 + end_date: 2024-03-31 + basis: tax + source_period_label: 2024 tax year +``` + +This differs from Chronicle's opening-year treatment of split UK labels such as +FY2024-25. + +## Geography and aggregation rulings + +These decisions implement the rulings recorded in +[issue #175](https://github.com/PolicyEngine/chronicle/issues/175): + +- Territorial authorities use `geography_level: local_authority` and + `geography_vintage: ta_2025`. +- MSD Work & Income regions use `geography_level: statistical_scope`, + `geography_vintage: msd_wi_region`, and stable `nz-wi-...` slug IDs. +- SA2 is deferred to wave 3, issue #178. When admitted, it will be a + first-class `sa2` geography rather than `statistical_scope`. +- Rent quartiles are not representable by Chronicle's current aggregation + vocabulary. Wave 1 records publisher medians as `aggregation: median` and + geometric means as `aggregation: mean`, `concept_relation: approximate`, + with evidence notes that explicitly identify the geometric mean. Quartiles + wait for a deliberate parameterized-percentile contract change. + +## Wave-1 ingestion ledger + +An unchecked row means that package work remains; it does not imply that the +official source is unavailable. Each completed row must pin the publisher +artifact and checksum, validate its source package, pass its country regression +tests, and record a verified `raw/nz/...` R2 URI. + +| Package from #176 | Artifact pinned | Package valid | `raw/nz` verified | Notes | +|---|---:|---:|---:|---| +| `stats_nz/subnational_population_estimates_2025` | [ ] | [ ] | [ ] | | +| `stats_nz/national_population_estimates_2025` | [ ] | [ ] | [ ] | | +| `stats_nz/census_2023_households_by_region` | [ ] | [ ] | [ ] | | +| `stats_nz/census_2023_family_type` | [ ] | [ ] | [ ] | | +| `stats_nz/census_2023_ethnicity_age_region` | [ ] | [ ] | [ ] | | +| `ird/taxable_income_distribution_2025` | [ ] | [ ] | [ ] | | +| `ird/wage_salary_distribution_2025` | [ ] | [ ] | [ ] | | +| `ird/working_for_families_statistics_sept_2025` | [ ] | [ ] | [ ] | | +| `ird/student_loan_statistics_march_2026` | [ ] | [ ] | [ ] | | +| `msd/benefit_fact_sheets_national_march_2026` | [ ] | [ ] | [ ] | | +| `msd/benefit_fact_sheets_supplementary_march_2026` | [ ] | [ ] | [ ] | | +| `msd/nzs_vp_fact_sheet_march_2026` | [ ] | [ ] | [ ] | | +| `msd/annual_report_benefit_expenses_2025` | [ ] | [ ] | [ ] | | +| `mbie/tenancy_bond_rents_tla_2026` | [ ] | [ ] | [ ] | | +| `stats_nz/qes_average_earnings_march_2026` | [ ] | [ ] | [ ] | | diff --git a/docs/storage-architecture.md b/docs/storage-architecture.md index 826ece4d..cf898933 100644 --- a/docs/storage-architecture.md +++ b/docs/storage-architecture.md @@ -46,35 +46,41 @@ Hosted tables mirror accepted build outputs and provide a shared query surface. ## Object Key Conventions -Raw source artifacts use the implemented content-addressed key shape: +New UK and New Zealand source artifacts use country-organized, +content-addressed keys: ```text -raw/{source_id}/{package_id}/{year}/{sha256}/{filename} +raw/{country}/{source_id}/{package_id}/{year}/{sha256}/{filename} ``` -For example: +For example, an IRD artifact uses: ```text -raw/irs_soi/soi-table-1-1/2023/842da11...aca17/23in11si.xls +raw/nz/ird/ird-working-for-families-statistics-sept-2025/2024/{sha256}/working-for-families-statistics---sept-2025.xlsx ``` -Derived build artifacts should use build-scoped keys so different builds can -coexist and be audited: +The implemented country segments are `nz` and `uk`. US objects deliberately +retain the legacy shape `raw/{source_id}/...`; migrating those keys requires a +separate consumer audit. The fetch and raw-publish commands infer the country +from the package publisher directory. Raw publication refuses to replace a +manifest-recorded key that disagrees with the inferred country path. + +New UK and New Zealand derived build artifacts use the same country segment +and build-scoped keys so different builds can coexist and be audited: ```text -derived/{source_id}/{package_id}/{year}/{build_id}/{artifact_name} +derived/{country}/{source_id}/{package_id}/{year}/{build_id}/{artifact_name} ``` Examples: ```text -derived/irs_soi/soi-table-1-1/2023/{build_id}/source_cells.jsonl -derived/bea/bea-nipa-pension-contributions/2022/{build_id}/source_rows.jsonl -derived/irs_soi/soi-table-1-1/2023/{build_id}/ledger.db -derived/irs_soi/soi-table-1-1/2023/{build_id}/reports/build_summary.json -derived/irs_soi/soi-table-1-1/2023/{build_id}/mirror/aggregate_facts.jsonl +derived/uk/ons/ons-mye-2024-uk/2024/{build_id}/source_cells.jsonl +derived/nz/ird/ird-working-for-families-statistics-sept-2025/2024/{build_id}/ledger.db ``` +Legacy US derived keys likewise remain `derived/{source_id}/...`. + Derived artifacts are reproducible and may be replaced by a new build, but a specific `{build_id}` path should be immutable once published. diff --git a/tests/test_chronicle_artifacts.py b/tests/test_chronicle_artifacts.py index 2728965b..f5b3c259 100644 --- a/tests/test_chronicle_artifacts.py +++ b/tests/test_chronicle_artifacts.py @@ -15,6 +15,7 @@ bootstrap_r2_buckets, build_r2_key, fetch_source_artifact, + infer_r2_country, infer_build_id, inventory_source_artifacts, publish_derived_artifacts, @@ -50,6 +51,55 @@ def test_build_derived_r2_key_is_build_scoped(): ) +@pytest.mark.parametrize( + ("source_id", "package_path", "expected_country"), + [ + ("ird", "db/data/ird/wff", "nz"), + ("ons", "db/data/ons/mye", "uk"), + ("irs_soi", "db/data/irs_soi/table_1_1", None), + ], +) +def test_infer_r2_country_uses_publisher_directory( + source_id, package_path, expected_country +): + assert ( + infer_r2_country(source_id=source_id, package_path=package_path) + == expected_country + ) + + +def test_country_aware_r2_keys_preserve_legacy_us_layout(): + nz_key = build_r2_key( + source_id="ird", + package_id="ird-working-for-families-statistics-sept-2025", + year=2024, + sha256="abc123", + filename="wff.xlsx", + ) + uk_key = build_derived_r2_key( + source_id="ons", + package_id="ons-mye-2024-uk", + year=2024, + build_id="ledger.build.v1:abc123", + artifact_name="facts.jsonl", + ) + + assert nz_key.startswith("raw/nz/ird/") + assert uk_key.startswith("derived/uk/ons/") + + +def test_country_aware_r2_prefix_rejects_wrong_country(): + with pytest.raises(ValueError, match="disagrees with publisher country"): + build_r2_key( + source_id="ird", + package_id="wff", + year=2024, + sha256="abc123", + filename="wff.xlsx", + prefix="raw/uk", + ) + + def test_fetch_source_artifact_writes_manifest_and_inventory(tmp_path): source = tmp_path / "source.xls" content = b"chronicle artifact fixture" @@ -160,6 +210,77 @@ def test_publish_source_artifacts_handles_label_year_entries(tmp_path): ) +def test_publish_source_artifacts_uses_country_for_each_manifest(tmp_path): + log = tmp_path / "wrangler.log" + wrangler = tmp_path / "wrangler" + wrangler.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$*\" >> {log}\necho ok\n") + wrangler.chmod(0o755) + for publisher, package_id, year in ( + ("ird", "ird-wff", 2024), + ("ons", "ons-mye", 2024), + ("irs_soi", "soi-table", 2023), + ): + output_dir = tmp_path / "data" / publisher / package_id + source = tmp_path / f"{publisher}.csv" + source.write_bytes(publisher.encode()) + fetch_source_artifact( + str(source), + source_id=publisher, + package_id=package_id, + year=year, + output_dir=output_dir, + ) + + report = publish_source_artifacts(tmp_path / "data", wrangler_command=str(wrangler)) + + assert report.valid + commands = log.read_text() + assert "ledger-raw/raw/nz/ird/ird-wff/2024/" in commands + assert "ledger-raw/raw/uk/ons/ons-mye/2024/" in commands + assert "ledger-raw/raw/irs_soi/soi-table/2023/" in commands + + +def test_publish_source_artifacts_refuses_stale_country_key(tmp_path): + output_dir = tmp_path / "data" / "ird" / "wff" + source = tmp_path / "wff.xlsx" + source.write_bytes(b"official WFF workbook") + fetch_source_artifact( + str(source), + source_id="ird", + package_id="ird-wff", + year=2024, + output_dir=output_dir, + ) + manifest_path = output_dir / "manifest.yaml" + manifest = yaml.safe_load(manifest_path.read_text()) + artifact = manifest["files"][2024] + artifact["storage"] = { + "r2": { + "provider": "r2", + "bucket": "ledger-raw", + "key": ( + f"raw/ird/ird-wff/2024/{artifact['sha256']}/{artifact['filename']}" + ), + } + } + manifest_path.write_text(yaml.safe_dump(manifest, sort_keys=False)) + log = tmp_path / "wrangler.log" + wrangler = tmp_path / "wrangler" + wrangler.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$*\" >> {log}\necho ok\n") + wrangler.chmod(0o755) + + report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) + + assert not report.valid + assert report.entries[0].upload is None + assert ( + report.entries[0] + .errors[0] + .startswith("recorded_r2_key_disagrees_with_country_prefix:") + ) + assert not log.exists() + + def test_inventory_source_artifacts_catches_checksum_mismatch(tmp_path): source = tmp_path / "source.xls" source.write_bytes(b"original") From ccc58d702d04ee9a08b4e3434a51759c2aa4acc5 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sat, 29 Aug 2026 19:46:24 -0400 Subject: [PATCH 2/3] Guard country routing before artifact writes (#175) Restrict publisher inference to canonical package roots, preserve cached bytes on rejected routes, and correct the stale quartile ruling using the existing quantile contract. Targeted schema, consumer, artifact, core, and import checks: 83 passed. --- chronicle/artifacts.py | 55 ++++++++++++++++++------------- docs/pe-nz-source-checklist.md | 13 +++++--- tests/test_chronicle_artifacts.py | 45 +++++++++++++++++++++++-- 3 files changed, 84 insertions(+), 29 deletions(-) diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py index c2b944fa..e019da98 100644 --- a/chronicle/artifacts.py +++ b/chronicle/artifacts.py @@ -395,13 +395,19 @@ def fetch_source_artifact( wrangler_command: str = "npx wrangler", ) -> ArtifactFetchReport: """Fetch/register a source artifact and optionally upload it to R2.""" + output = Path(output_dir) + resolved_r2_prefix = resolve_r2_prefix( + prefix=r2_prefix, + default_prefix=DEFAULT_R2_PREFIX, + source_id=source_id, + package_path=output, + ) fetched_at = datetime.now(UTC).replace(microsecond=0).isoformat() content, inferred_filename = _read_artifact(source_url) artifact_filename = filename or inferred_filename if not artifact_filename: raise ValueError("Could not infer artifact filename; pass --filename.") - output = Path(output_dir) output.mkdir(parents=True, exist_ok=True) local_path = output / artifact_filename local_path.write_bytes(content) @@ -410,12 +416,6 @@ def fetch_source_artifact( size_bytes = len(content) manifest_path = output / "manifest.yaml" - resolved_r2_prefix = resolve_r2_prefix( - prefix=r2_prefix, - default_prefix=DEFAULT_R2_PREFIX, - source_id=source_id, - package_path=output, - ) r2_location = ArtifactStorageLocation( provider="r2", bucket=r2_bucket, @@ -426,6 +426,7 @@ def fetch_source_artifact( sha256=sha256, filename=artifact_filename, prefix=resolved_r2_prefix, + package_path=output, ), ) r2_upload = None @@ -822,19 +823,27 @@ def infer_r2_country( fallback for low-level callers that do not have a package path, including derived-artifact publication. """ - path_countries: set[str] = set() + publisher = None if package_path is not None: - path_parts = {part.lower() for part in Path(package_path).parts} - path_countries = { + path_parts = tuple(part.lower() for part in Path(package_path).parts) + # Match the innermost canonical package root, not arbitrary ancestor + # directory names such as /work/ons/chronicle/db/data/irs_soi/.... + for index in range(len(path_parts) - 1, -1, -1): + if path_parts[index : index + 2] == ("db", "data"): + if index + 2 < len(path_parts): + publisher = path_parts[index + 2] + break + elif path_parts[index] == "packages" and index + 1 < len(path_parts): + publisher = path_parts[index + 1] + break + path_country = next( + ( country for country, publishers in R2_COUNTRY_PUBLISHERS.items() - if path_parts.intersection(publishers) - } - if len(path_countries) > 1: - raise ValueError( - "package path contains publishers from multiple countries: " - f"{package_path}" - ) + if publisher in publishers + ), + None, + ) source_countries: set[str] = set() normalized_source_id = (source_id or "").lower() @@ -852,14 +861,13 @@ def infer_r2_country( if len(source_countries) > 1: raise ValueError(f"source_id maps to multiple R2 countries: {source_id}") - if path_countries and source_countries and path_countries != source_countries: + source_country = next(iter(source_countries)) if source_countries else None + if publisher is not None and source_country and source_country != path_country: raise ValueError( "package publisher directory and source_id map to different " - f"countries: path={sorted(path_countries)}, " - f"source_id={sorted(source_countries)}" + f"countries: publisher={publisher!r}, source_id={source_id!r}" ) - countries = path_countries or source_countries - return next(iter(countries)) if countries else None + return path_country if publisher is not None else source_country def resolve_r2_prefix( @@ -894,6 +902,7 @@ def build_r2_key( sha256: str, filename: str, prefix: str | None = None, + package_path: str | Path | None = None, ) -> str: """Build the canonical immutable R2 key for a raw source artifact. @@ -904,6 +913,7 @@ def build_r2_key( prefix=prefix, default_prefix=DEFAULT_R2_PREFIX, source_id=source_id, + package_path=package_path, ) return posixpath.join( resolved_prefix, @@ -1132,6 +1142,7 @@ def _publish_raw_manifest_entry( sha256=sha256_actual or "", filename=filename, prefix=r2_prefix, + package_path=manifest_path, ), ) storage = spec.get("storage") if isinstance(spec.get("storage"), dict) else {} diff --git a/docs/pe-nz-source-checklist.md b/docs/pe-nz-source-checklist.md index ab98924f..c0b362a5 100644 --- a/docs/pe-nz-source-checklist.md +++ b/docs/pe-nz-source-checklist.md @@ -35,11 +35,14 @@ These decisions implement the rulings recorded in `geography_vintage: msd_wi_region`, and stable `nz-wi-...` slug IDs. - SA2 is deferred to wave 3, issue #178. When admitted, it will be a first-class `sa2` geography rather than `statistical_scope`. -- Rent quartiles are not representable by Chronicle's current aggregation - vocabulary. Wave 1 records publisher medians as `aggregation: median` and - geometric means as `aggregation: mean`, `concept_relation: approximate`, - with evidence notes that explicitly identify the geometric mean. Quartiles - wait for a deliberate parameterized-percentile contract change. +- The original rent-quartile blocker in #175 predates Eurostat #168. + Chronicle now supports `aggregation: quantile`; publisher quartile cut-points + can use it with the percentile label or code carried in explicit constraints + and source evidence, following `eurostat/ilc_di01`. No new aggregation name is + needed. This removes a vocabulary blocker without expanding #176's package + scope. Publisher medians use `aggregation: median`; geometric means use + `aggregation: mean`, `concept_relation: approximate`, and evidence notes that + explicitly identify the geometric mean. ## Wave-1 ingestion ledger diff --git a/tests/test_chronicle_artifacts.py b/tests/test_chronicle_artifacts.py index f5b3c259..ccab797d 100644 --- a/tests/test_chronicle_artifacts.py +++ b/tests/test_chronicle_artifacts.py @@ -57,6 +57,10 @@ def test_build_derived_r2_key_is_build_scoped(): ("ird", "db/data/ird/wff", "nz"), ("ons", "db/data/ons/mye", "uk"), ("irs_soi", "db/data/irs_soi/table_1_1", None), + ("irs_soi", "/work/ons/chronicle/db/data/irs_soi/table_1_1", None), + ("ird", "/work/ons/chronicle/db/data/ird/wff", "nz"), + ("ird", "/work/packages/ons/chronicle/db/data/ird/wff", "nz"), + ("ird", "/work/ons/chronicle/packages/ird/wff", "nz"), ], ) def test_infer_r2_country_uses_publisher_directory( @@ -137,6 +141,42 @@ def test_fetch_source_artifact_writes_manifest_and_inventory(tmp_path): } +def test_fetch_rejects_wrong_country_before_reading_or_overwriting_cache( + tmp_path, monkeypatch +): + source = tmp_path / "source.xlsx" + source.write_bytes(b"original official artifact") + output_dir = tmp_path / "db" / "data" / "ird" / "wff" + fetch_source_artifact( + str(source), + source_id="ird", + package_id="ird-wff", + year=2024, + output_dir=output_dir, + ) + artifact_path = output_dir / source.name + manifest_path = output_dir / "manifest.yaml" + original_artifact = artifact_path.read_bytes() + original_manifest = manifest_path.read_bytes() + + def unexpected_read(_source_url): + raise AssertionError("A rejected route must not read the source artifact") + + monkeypatch.setattr("chronicle.artifacts._read_artifact", unexpected_read) + with pytest.raises(ValueError, match="disagrees with publisher country"): + fetch_source_artifact( + str(source), + source_id="ird", + package_id="ird-wff", + year=2024, + output_dir=output_dir, + r2_prefix="raw/uk", + ) + + assert artifact_path.read_bytes() == original_artifact + assert manifest_path.read_bytes() == original_manifest + + def test_publish_source_artifacts_uploads_manifest_entries(tmp_path): output_dir = tmp_path / "data" / "irs_soi" / "table_1_2" source = tmp_path / "source.xls" @@ -211,6 +251,7 @@ def test_publish_source_artifacts_handles_label_year_entries(tmp_path): def test_publish_source_artifacts_uses_country_for_each_manifest(tmp_path): + data_root = tmp_path / "ons" / "chronicle" / "db" / "data" log = tmp_path / "wrangler.log" wrangler = tmp_path / "wrangler" wrangler.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$*\" >> {log}\necho ok\n") @@ -220,7 +261,7 @@ def test_publish_source_artifacts_uses_country_for_each_manifest(tmp_path): ("ons", "ons-mye", 2024), ("irs_soi", "soi-table", 2023), ): - output_dir = tmp_path / "data" / publisher / package_id + output_dir = data_root / publisher / package_id source = tmp_path / f"{publisher}.csv" source.write_bytes(publisher.encode()) fetch_source_artifact( @@ -231,7 +272,7 @@ def test_publish_source_artifacts_uses_country_for_each_manifest(tmp_path): output_dir=output_dir, ) - report = publish_source_artifacts(tmp_path / "data", wrangler_command=str(wrangler)) + report = publish_source_artifacts(data_root, wrangler_command=str(wrangler)) assert report.valid commands = log.read_text() From 544b26e96545f829a6afe98ec7d094b04b98c76b Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sat, 29 Aug 2026 19:52:41 -0400 Subject: [PATCH 3/3] Require publisher and package structure for R2 routing (#175) --- chronicle/artifacts.py | 9 +++++---- tests/test_chronicle_artifacts.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 4 deletions(-) diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py index e019da98..d9620761 100644 --- a/chronicle/artifacts.py +++ b/chronicle/artifacts.py @@ -826,14 +826,15 @@ def infer_r2_country( publisher = None if package_path is not None: path_parts = tuple(part.lower() for part in Path(package_path).parts) - # Match the innermost canonical package root, not arbitrary ancestor - # directory names such as /work/ons/chronicle/db/data/irs_soi/.... + # Match the innermost canonical root with both publisher and package + # directories. A package itself may be named "packages"; the following + # manifest filename must not then be mistaken for a publisher. for index in range(len(path_parts) - 1, -1, -1): if path_parts[index : index + 2] == ("db", "data"): - if index + 2 < len(path_parts): + if index + 3 < len(path_parts): publisher = path_parts[index + 2] break - elif path_parts[index] == "packages" and index + 1 < len(path_parts): + elif path_parts[index] == "packages" and index + 2 < len(path_parts): publisher = path_parts[index + 1] break path_country = next( diff --git a/tests/test_chronicle_artifacts.py b/tests/test_chronicle_artifacts.py index ccab797d..e08beabc 100644 --- a/tests/test_chronicle_artifacts.py +++ b/tests/test_chronicle_artifacts.py @@ -61,6 +61,10 @@ def test_build_derived_r2_key_is_build_scoped(): ("ird", "/work/ons/chronicle/db/data/ird/wff", "nz"), ("ird", "/work/packages/ons/chronicle/db/data/ird/wff", "nz"), ("ird", "/work/ons/chronicle/packages/ird/wff", "nz"), + ("ird", "/repo/db/data/ird/packages", "nz"), + ("ird", "/repo/db/data/ird/packages/manifest.yaml", "nz"), + ("ird", "/repo/packages/ird/packages/source_package.yaml", "nz"), + ("irs_soi", "/repo/db/data/irs_soi/packages/manifest.yaml", None), ], ) def test_infer_r2_country_uses_publisher_directory( @@ -322,6 +326,32 @@ def test_publish_source_artifacts_refuses_stale_country_key(tmp_path): assert not log.exists() +def test_publish_source_artifacts_accepts_package_named_packages(tmp_path): + output_dir = tmp_path / "db" / "data" / "ird" / "packages" + source = tmp_path / "wff.xlsx" + source.write_bytes(b"official WFF workbook") + fetched = fetch_source_artifact( + str(source), + source_id="ird", + package_id="packages", + year=2024, + output_dir=output_dir, + ) + wrangler = tmp_path / "wrangler" + wrangler.write_text("#!/bin/sh\necho ok\n") + wrangler.chmod(0o755) + + published = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) + + assert fetched.valid + assert published.valid + assert published.counts["uploaded_count"] == 1 + manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text()) + assert manifest["files"][2024]["storage"]["r2"]["key"].startswith( + "raw/nz/ird/packages/2024/" + ) + + def test_inventory_source_artifacts_catches_checksum_mismatch(tmp_path): source = tmp_path / "source.xls" source.write_bytes(b"original")