fix: add precondition assertions in extract pipeline

Assert landing_dir.is_dir() and year_month format (YYYY/MM) at the
entry point of each extract function — turning silent wrong-path bugs
into immediate AssertionError with a descriptive message.

Files changed:
- playtomic_availability.py: assert in _load_tenant_ids(), extract(),
  extract_recheck()
- eurostat.py: assert in extract()

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-02-24 20:42:11 +01:00
parent dd9ffd6d27
commit ad48f23cfc
2 changed files with 7 additions and 0 deletions

View File

@@ -180,6 +180,8 @@ def extract(
session: niquests.Session, session: niquests.Session,
) -> dict: ) -> dict:
"""Fetch all Eurostat datasets. Returns run metrics.""" """Fetch all Eurostat datasets. Returns run metrics."""
assert landing_dir.is_dir(), f"landing_dir must exist: {landing_dir}"
assert "/" in year_month and len(year_month) == 7, f"year_month must be YYYY/MM: {year_month!r}"
year, month = year_month.split("/") year, month = year_month.split("/")
files_written = 0 files_written = 0
files_skipped = 0 files_skipped = 0

View File

@@ -59,6 +59,7 @@ _thread_local = threading.local()
def _load_tenant_ids(landing_dir: Path) -> list[str]: def _load_tenant_ids(landing_dir: Path) -> list[str]:
"""Read tenant IDs from the most recent tenants.json.gz file.""" """Read tenant IDs from the most recent tenants.json.gz file."""
assert landing_dir.is_dir(), f"landing_dir must exist: {landing_dir}"
playtomic_dir = landing_dir / "playtomic" playtomic_dir = landing_dir / "playtomic"
if not playtomic_dir.exists(): if not playtomic_dir.exists():
return [] return []
@@ -243,6 +244,8 @@ def extract(
session: niquests.Session, session: niquests.Session,
) -> dict: ) -> dict:
"""Fetch next-day availability for all known Playtomic venues.""" """Fetch next-day availability for all known Playtomic venues."""
assert landing_dir.is_dir(), f"landing_dir must exist: {landing_dir}"
assert "/" in year_month and len(year_month) == 7, f"year_month must be YYYY/MM: {year_month!r}"
tenant_ids = _load_tenant_ids(landing_dir) tenant_ids = _load_tenant_ids(landing_dir)
if not tenant_ids: if not tenant_ids:
logger.warning("No tenant IDs found — run extract-playtomic-tenants first") logger.warning("No tenant IDs found — run extract-playtomic-tenants first")
@@ -385,6 +388,8 @@ def extract_recheck(
session: niquests.Session, session: niquests.Session,
) -> dict: ) -> dict:
"""Re-query venues with slots starting soon for accurate occupancy data.""" """Re-query venues with slots starting soon for accurate occupancy data."""
assert landing_dir.is_dir(), f"landing_dir must exist: {landing_dir}"
assert "/" in year_month and len(year_month) == 7, f"year_month must be YYYY/MM: {year_month!r}"
now = datetime.now(UTC) now = datetime.now(UTC)
target_date = now.strftime("%Y-%m-%d") target_date = now.strftime("%Y-%m-%d")