Skip to content

yggdrasil.enums.media_type

media_type

Format + codec wrapper resolution.

Where :class:MimeType resolves a single MIME identifier (string, extension, magic bytes), :class:MediaType resolves the combination of an inner format and an outer wrapper codec — what you actually need to dispatch reads/writes.

Two-stage sniffing

For bytes-shaped and file-like inputs the resolution is two-stage. The outer magic identifies the wrapper (b"\x1f\x8b..." → GZIP). When the wrapper is a codec we then peek a head-window of the decompressed payload and sniff that — so a gzip-wrapped parquet returns MediaType(PARQUET, codec=GZIP) rather than the lossy MediaType(OCTET_STREAM, codec=GZIP) that comes from sniffing the outer mime alone.

Two entry points:

  • :meth:MediaType.from_magic — bytes / bytearray / memoryview. No cursor concerns; sniffs in-memory.
  • :meth:MediaType.from_io — file-like with read + seek. Seeks the source freely during sniffing (one read at position 0 for the outer head; if a codec is identified, another seek to 0 to drive a streaming decompressor for the inner head). The caller's cursor is captured on entry and restored on exit, so the IO is left exactly where it came in regardless of how much internal seeking happened.

Both entry points fall back to MediaType(OCTET_STREAM, codec=<wrapper>) when the wrapper is identified but the inner sniff fails — losing nothing the caller already had.

MediaType dataclass 

MediaType(mime_type: MimeType, codec: Codec | None = None)

from_many classmethod 

from_many(
    mime_types: Iterable[MimeType], default: "MediaType" = ...
) -> "MediaType"

Compose a MediaType from an ordered mime list (e.g. URL extensions).

Two conventions land here:

  • trades.csv.zst → ["csv", "zst"]: the codec is the outer wrapper (you have to unzst before you can parse csv) → MediaType(CSV, codec=ZSTD).
  • part-xxx.zstd.parquet → ["zstd", "parquet"]: the format is the outer wrapper and the codec is the parquet page-codec hint baked into the file. Parquet handles the decompression internally → MediaType(PARQUET) with no outer codec; setting one would route the read through a decompressor that doesn't belong on this byte stream.

Order matters: the last mime decides. Last-is-codec promotes the codec to the wrapper slot; last-is-format keeps the format alone and drops earlier codec hints.

from_magic classmethod 

from_magic(
    buf: Union[bytes, bytearray, memoryview, IO[bytes]],
    *,
    default: "MediaType" = ...
) -> "MediaType"

Sniff a :class:MediaType from raw bytes.

Two-stage: outer magic identifies the wrapper; if the wrapper is a codec, decompress a head-window and sniff the inner format. See module docstring for the full story.

Accepts bytes / bytearray / memoryview directly. For convenience, also accepts an IO[bytes] — in that case it delegates to :meth:from_io, which manages cursor save/restore.

:param default: Fallback when the outer sniff finds nothing. ... (default) raises :class:ValueError; any other value (including None) is returned as-is.

from_io classmethod 

from_io(io_obj: IO[bytes], *, default: 'MediaType' = ...) -> 'MediaType'

Sniff a :class:MediaType from a file-like, seeking freely.

Captures the cursor on entry and restores it on exit, so the caller's stream position is unaffected. Inside the call we seek freely:

  1. seek(0) and read the first :data:_OUTER_PEEK bytes for outer-magic resolution.
  2. If the outer mime is a codec, seek(0) again and drive the codec's streaming decompressor directly to read the first :data:_INNER_PEEK bytes of decoded payload.
  3. Sniff the decoded head for the inner mime.

The codec resolution is one dict lookup (:meth:Codec.from_mime) — the outer magic loop already landed on the registered :class:MimeType singleton, so we reuse it as the key without re-resolving. For streaming codecs (gzip / zstd / lz4 / bz2 / xz / zlib / lzma) the decompressor is opened directly on the IO via :meth:Codec._open_decompress_reader — no IO wrap, no full-buffer materialization. Non-streaming codecs (snappy, brotli) fall back to reading the compressed body and calling :meth:Codec.decompress_bytes.

Decompression errors during the inner sniff are swallowed — the worst case is returning a less-specific MediaType (the outer codec is preserved). A separate caller that wants to validate decompressibility should call :meth:Codec.decompress directly.

:param io_obj: Any file-like with read + seek (stdlib io.BytesIO, an open file, our own IO, etc.). :param default: Fallback when the outer sniff finds nothing. ... (default) raises :class:ValueError; any other value (including None) is returned as-is.