How to Use the Magika Python API with File Streams: A Complete Guide

The Magika.identify_stream() method accepts any readable binary stream—whether a file object, io.BytesIO, or custom BinaryIO implementation—and returns a content type prediction while automatically preserving the original stream position.

The google/magika repository provides a machine learning-powered file type identification library that processes content via an ONNX model. When you need to classify files that are already opened in memory or streamed from network sources, the Python API's stream-based interface eliminates the need to write temporary files to disk. This guide demonstrates how to leverage identify_stream with various binary stream types while ensuring cursor integrity, based on the implementation in python/src/magika/magika.py.

Using identify_stream with File Objects

The most common use case involves passing an already-opened file handle to the identify_stream method. In python/src/magika/magika.py at lines 177-211, the Magika class defines this entry point, which validates that the provided object implements io.IOBase, is readable, and is opened in binary mode.

from magika import Magika

magika = Magika()

# Open a file in binary mode and pass the file object directly

with open("sample.pdf", "rb") as f:
    result = magika.identify_stream(f)

print("Detected type:", result.prediction.output.label)

# → Detected type: PDF

The method returns a MagikaResult object containing the prediction metadata. As demonstrated in the test suite at python/tests/test_magika_python_module.py lines 12-18, this pattern works for any file path opened with "rb" mode.

Processing In-Memory BytesIO Streams

For data already residing in memory—such as uploaded file contents or generated payloads—you can wrap bytes in io.BytesIO and pass the resulting stream directly. The implementation treats BytesIO identically to file objects because both satisfy the BinaryIO protocol.

import io
from magika import Magika

magika = Magika()

binary_content = b"%PDF-1.4\n%âãÏÓ\n..."
stream = io.BytesIO(binary_content)

result = magika.identify_stream(stream)
print(result.prediction.output.label)   # PDF

Internally, Magika.identify_bytes creates exactly this type of BytesIO wrapper before delegating to the stream processing logic, ensuring consistent behavior between byte arrays and stream objects.

Supporting Custom BinaryIO Implementations

The API accepts any object implementing the BinaryIO abstract base class. Your custom class must provide read(), seek(), tell(), and readable() methods. The identify_stream method wraps your object with a Seekable helper (defined in python/src/magika/types/seekable.py) that enables random-access reads required by the ONNX inference engine.

import io
from typing import BinaryIO
from magika import Magika

class SimpleChunkedReader(BinaryIO):
    """A minimal BinaryIO that reads from a pre-loaded bytes object."""
    def __init__(self, data: bytes):
        self._buffer = io.BytesIO(data)

    def read(self, size=-1):
        return self._buffer.read(size)

    def seek(self, offset, whence=io.SEEK_SET):
        return self._buffer.seek(offset, whence)

    def tell(self):
        return self._buffer.tell()

    def readable(self):
        return True

    # Required by BinaryIO

    def close(self):
        self._buffer.close()

    def closed(self):
        return self._buffer.closed

# Usage

data = b"\x89PNG\r\n\x1a\n..."   # PNG file header

custom_stream: BinaryIO = SimpleChunkedReader(data)

magika = Magika()
result = magika.identify_stream(custom_stream)
print(result.prediction.output.label)   # PNG

How Stream Position Preservation Works

A critical feature of identify_stream is that it guarantees the caller's stream position remains unchanged after classification. The implementation saves the current cursor position using stream.tell() before processing, then restores it in a finally block regardless of success or failure.

import io
from magika import Magika

magika = Magika()
content = b"A" * 1000
stream = io.BytesIO(content)

# Move cursor to a non-zero offset

stream.seek(42)
original_pos = stream.tell()

result = magika.identify_stream(stream)

assert stream.tell() == original_pos, "Stream position was altered!"
print("Position preserved; detected:", result.prediction.output.label)

This behavior is verified in python/tests/test_magika_python_module.py by the test_magika_module_identify_stream_does_not_alter_position test case. Under the hood, the Seekable wrapper determines the total stream size by seeking to the end once, then uses read_at(offset, size) methods to feed specific byte windows to the model without modifying the original stream's state.

Summary

• identify_stream accepts any BinaryIO implementation, including file objects, BytesIO, and custom streams.
• Position preservation is automatic: the method saves and restores the cursor offset in python/src/magika/magika.py using a finally block.
• Protocol requirements are minimal: your stream needs read(), seek(), tell(), and readable() methods.
• Random access is handled via the internal Seekable wrapper in python/src/magika/types/seekable.py, which enables the ONNX model to extract specific byte ranges without sequential reading.

Frequently Asked Questions

Can I use identify_stream with text-mode file objects?

No. The method explicitly checks that the stream is opened in binary mode (or implements the binary interface) and will raise an error if passed a text-mode file. You must open files with "rb" mode or use io.BytesIO for in-memory text encoded as bytes.

What happens to the file cursor after identification?

The cursor returns to exactly where it was before the call. The implementation records stream.tell() before inference and restores that position after the Seekable wrapper finishes processing, even if an exception occurs during classification.

Does identify_stream load the entire file into memory?

No. The Seekable wrapper only reads specific byte ranges required by the model (typically small windows from the beginning, middle, and end of the file) using random-access seeks. This makes it safe to use with large files without consuming excessive memory.

What is the minimum interface my custom stream class must implement?

Your class must satisfy the BinaryIO protocol: implement read(), seek(), tell(), and readable() methods. The identify_stream method validates these capabilities before wrapping the object in the Seekable helper that the inference engine uses.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}