Create utils.py

src/utils.py

@@ -0,0 +1,44 @@
+# src/utils.py
+import os
+import sqlite3
+import pathlib
+from datetime import datetime
+import streamlit as st
+
+APP_DIR = pathlib.Path(__file__).resolve().parent.parent
+DB_PATH = APP_DIR / "olist.sqlite"
+
+# Make local writable dirs for caches
+os.environ.setdefault("HOME", str(APP_DIR))
+os.environ.setdefault("XDG_CACHE_HOME", str(APP_DIR / ".cache"))
+os.environ.setdefault("MPLCONFIGDIR", str(APP_DIR / ".mplconfig"))
+os.makedirs(APP_DIR / ".cache", exist_ok=True)
+os.makedirs(APP_DIR / ".mplconfig", exist_ok=True)
+
+def ensure_db_present():
+    """Ensure olist.sqlite is present at DB_PATH (copy from repo if needed)."""
+    if DB_PATH.exists():
+        return
+    # repo copy
+    repo_db = APP_DIR / "olist.sqlite"
+    if repo_db.exists():
+        repo_db.replace(DB_PATH)
+        return
+    # optional: download from HF Hub if configured
+    try:
+        from huggingface_hub import hf_hub_download
+        repo_id = os.getenv("HF_DB_REPO")
+        filename = os.getenv("HF_DB_FILE", "olist.sqlite")
+        token = os.getenv("HF_TOKEN")
+        if repo_id:
+            local = hf_hub_download(repo_id=repo_id, filename=filename, token=token)
+            pathlib.Path(local).replace(DB_PATH)
+    except Exception:
+        pass
+
+ensure_db_present()
+
+def get_connection() -> sqlite3.Connection:
+    uri = f"file:{DB_PATH}?mode=rwc"
+    conn = sqlite3.connect(uri, uri=True, check_same_thread=False)
+    return conn