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

> Learn to use the Magika Python API with file streams easily. Discover how Magika identify stream preserves stream position while predicting content types.

- Repository: [Google/magika](https://github.com/google/magika)
- Tags: how-to-guide
- Published: 2026-04-16

---

**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`](https://github.com/google/magika/blob/main/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`](https://github.com/google/magika/blob/main/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.

```python
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`](https://github.com/google/magika/blob/main/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.

```python
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`](https://github.com/google/magika/blob/main/python/src/magika/types/seekable.py)) that enables random-access reads required by the ONNX inference engine.

```python
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.

```python
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`](https://github.com/google/magika/blob/main/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`](https://github.com/google/magika/blob/main/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`](https://github.com/google/magika/blob/main/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.