INNER CODE UNIT · Python

api_enrich_status

Nuclear-Marmalade/dataforge · forge/dashboard/app.py:520

async def api_enrich_status():
    """Return enrichment status as JSON (polled by HTMX every 3s)."""
    with _enrichment_lock:
        stats_copy = dict(_enrichment_stats)
    return JSONResponse({"stats": stats_copy})


MAX_UPLOAD_SIZE = 100 * 1024 * 1024  # 100 MB


async def _save_upload_to_temp(file: UploadFile) -> tuple:
    """Stream upload to temp file. Returns (tmp_path, tmp_dir) or raises."""
    tmp_dir = tempfile.mkdtemp(prefix="forge_upload_")
    safe_name = secrets.token_hex(8) + ".csv"
    tmp_path = os.path.join(tmp_dir, safe_name)
    total = 0
    with open(tmp_path, "wb") as f:
        while True:

View source record →