angee.storage.uploads

ORM-free byte helpers shared by the storage upload flow.

The upload behavior lives on the model (File.objects.draft, File.receive_bytes / File.finalize); this module holds only the token constants and byte-level helpers they compose with — stream hashing, MIME detection, and the capped body reader. Keeping it free of any model import is what lets models.py import from here without a cycle.

FALLBACK_MIME

MIME used when libmagic yields nothing; also the guaranteed catalogue row.

UPLOAD_TOKEN_MAX_AGE

Seconds a proxy upload token stays valid.

UPLOAD_TOKEN_SALT

Signing salt namespacing proxy upload tokens.

UPLOAD_TOKEN_HEADER

Request header carrying the proxy upload token.

DOWNLOAD_TOKEN_MAX_AGE

Seconds a proxy download token stays valid.

DOWNLOAD_TOKEN_SALT

Signing salt namespacing proxy download tokens.

DOWNLOAD_TOKEN_HEADER

Request header carrying the proxy download token.

PROXY_CHUNK_SIZE

Bytes per read while streaming a proxied body into the backend.

MIME_SNIFF_BYTES

Head bytes captured during finalize hashing for MIME detection.

sha256_stream

def sha256_stream(reader: BinaryIO,
                  *,
                  capture_head: int = 0) -> tuple[str, int, bytes]

Stream-hash a binary reader without materializing it.

Returns (hex_digest, total_bytes, head_bytes); head_bytes is the first capture_head bytes so callers can sniff MIME without a second read.

detect_mime

def detect_mime(payload: bytes, filename: str = "") -> str

Detect a MIME type for a stored object.

libmagic sniffs the head bytes and is authoritative when it recognises the content. It does not know every format (e.g. HEIC on older magic databases), so when it yields nothing we fall back to the filename extension via the stdlib mimetypes registry, so the row still carries a useful type instead of the generic catch-all.

BodyTooLarge

class BodyTooLarge(Exception)

Internal sentinel raised by :class:CappedReader when the cap trips.

CappedReader

class CappedReader()

File-like wrapper that aborts once more than max_bytes are read.

Streams a request body into Storage.save without materializing it; Django's File wrapper only needs read(size). Overflow raises the private :class:BodyTooLarge sentinel so the caller can clean up the partial backend object before answering with an API error.

__init__

def __init__(reader: BinaryIO, *, max_bytes: int) -> None

Wrap reader with a hard byte cap.

read

def read(size: int = -1) -> bytes

Read up to size bytes, raising :class:BodyTooLarge on overflow.

close

def close() -> None

Close the wrapped reader when it supports closing.