from __future__ import annotations import uuid from pathlib import Path PARTIAL_NAME = "upload.partial" def new_session_dir(base: Path) -> tuple[str, Path]: session_id = uuid.uuid4().hex d = base / session_id d.mkdir(parents=True, exist_ok=True) return session_id, d def partial_path(session_dir: Path) -> Path: return session_dir / PARTIAL_NAME def write_chunk(session_dir: Path, data: bytes, offset: int) -> int: p = partial_path(session_dir) if offset < 0: raise ValueError("offset must be >= 0") current = p.stat().st_size if p.is_file() else 0 if offset != current: raise ValueError( f"Offset mismatch: expected {current} for append-only upload, got {offset}" ) with open(p, "ab") as f: f.write(data) return offset + len(data) def finalize_rename(session_dir: Path, final_name: str) -> Path: src = partial_path(session_dir) if not src.is_file() or src.stat().st_size == 0: raise FileNotFoundError("No uploaded data to finalize") dst = session_dir / final_name if dst.exists(): dst.unlink() src.rename(dst) return dst