check-metadata.py

import json
import sys
from datetime import datetime, timezone
from functools import cache
from pathlib import Path

import requests


def get_agency_url(agency: str):
    path = Path("./metadata.json")
    if not path.exists():
        raise RuntimeError("Metadata file not found")

    config = json.loads(path.read_text())
    return config[agency]


@cache
def get_metadata(url: str):
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    return response.json()


def _check_metadata_ts(url: str, ts_field: str):
    now = datetime.now(tz=timezone.utc)
    data = get_metadata(url)
    ts = data["db"][ts_field]
    timestamp = datetime.fromisoformat(ts)

    if not all((timestamp.year == now.year, timestamp.month == now.month, timestamp.day == now.day)):
        raise RuntimeError(f"Timestamp mismatch in {ts_field}: {ts}")


def check_metadata_file_ts(url: str):
    _check_metadata_ts(url, "file_ts")


def check_metadata_load_ts(url: str):
    _check_metadata_ts(url, "load_ts")


def check_metadata_users(url: str):
    data = get_metadata(url)
    users = data["db"]["users"]

    if users < 1:
        raise RuntimeError("Database has no users")


def check_metadata_eligibility(url: str):
    data = get_metadata(url)
    eligibility = data["db"]["eligibility"]

    if len(eligibility) < 1:
        raise RuntimeError("Database has no eligibility")


if __name__ == "__main__":
    args = sys.argv
    if len(args) < 2:
        raise RuntimeError("Usage: check-metadata AGENCY")

    agency = args[1]
    url = get_agency_url(agency)
    check_metadata_file_ts(url)
    check_metadata_load_ts(url)
    check_metadata_users(url)
    check_metadata_eligibility(url)