feat - sync geoip

src/leaderboard_analytics/geoip_database.py

@@ -0,0 +1,36 @@
+import gzip
+import shutil
+import tempfile
+from pathlib import Path
+from urllib.request import urlopen
+
+DEFAULT_GEOIP_DATABASE_URL = (
+    "https://cdn.jsdelivr.net/npm/geolite2-country/GeoLite2-Country.mmdb.gz"
+)
+
+
+def ensure_geoip_database(
+    database_path: str | Path,
+    source_url: str = DEFAULT_GEOIP_DATABASE_URL,
+    *,
+    auto_download: bool = True,
+    timeout: float = 30.0,
+) -> Path:
+    target_path = Path(database_path)
+    if target_path.exists() or not auto_download:
+        return target_path
+
+    target_path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        prefix=f"{target_path.name}.",
+        suffix=".tmp",
+        dir=target_path.parent,
+        delete=False,
+    ) as temp_file:
+        temp_path = Path(temp_file.name)
+        with urlopen(source_url, timeout=timeout) as response:
+            with gzip.GzipFile(fileobj=response) as gzip_file:
+                shutil.copyfileobj(gzip_file, temp_file)
+
+    temp_path.replace(target_path)
+    return target_path

tests/test_geoip_database.py

@@ -0,0 +1,29 @@
+import gzip
+
+from leaderboard_analytics.geoip_database import ensure_geoip_database
+
+
+def test_ensure_geoip_database_downloads_and_decompresses_gzip(tmp_path) -> None:
+    source = tmp_path / "GeoLite2-Country.mmdb.gz"
+    target = tmp_path / "GeoLite2-Country.mmdb"
+    expected_bytes = b"fake-mmdb-bytes"
+
+    with gzip.open(source, "wb") as gzip_file:
+        gzip_file.write(expected_bytes)
+
+    result = ensure_geoip_database(target, source.as_uri())
+
+    assert result == target
+    assert target.read_bytes() == expected_bytes
+
+
+def test_ensure_geoip_database_keeps_existing_file(tmp_path) -> None:
+    source = tmp_path / "missing.mmdb.gz"
+    target = tmp_path / "GeoLite2-Country.mmdb"
+    expected_bytes = b"existing-mmdb-bytes"
+    target.write_bytes(expected_bytes)
+
+    result = ensure_geoip_database(target, source.as_uri())
+
+    assert result == target
+    assert target.read_bytes() == expected_bytes