INNER CODE UNIT · Python

_save_upload_to_temp

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

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:
            chunk = await file.read(8192)
            if not chunk:
                break
            total += len(chunk)
            if total > MAX_UPLOAD_SIZE:
                f.close()
                os.unlink(tmp_path)
                os.rmdir(tmp_dir)
                raise ValueError("File too large")
            f.write(chunk)

View source record →