duck.utils.fileio

FileIOStream module.

Provides both synchronous and asynchronous file streaming interfaces. Ideal for efficient reading of large files using chunked reads and supporting standard seek, tell, and close operations in both environments.

Methods that do not need to be async: Even in async context, the below methods don’t necessarily need to be async:

1. open - Time complexity is O(1)

2. seek - Time complexity is O(1)

3. tell - Time complexity is O(1)

In async context, only read, write, and close need to be asynchronous.

Caching: Read results are cached in a shared LRU InMemoryCache keyed by filepath:pos:size. Each unique (file, offset, length) triple has its own independent cache slot, so reads from different positions are all cache-warm without interfering with each other — beneficial in environments that seek frequently. Stale entries (detected via mtime) are evicted on access. On write, overlapping cache entries are patched in memory without a disk round-trip: entries fully covered by the write are spliced directly; entries whose content is entirely inside the write range are reconstructed in place; boundary-partial overlaps are evicted. The exact slice just written is always stored in a new cache entry.

Events: Hooks can be attached to on_read and on_write events via hook(). Each hook receives (stream, data, byte_count) and can be a plain callable or an async coroutine function. Async hooks on a sync stream are scheduled fire-and-forget on the running event loop when one is available.

Example:

stream = FileIOStream("data.bin", open_now=True)

def log_read(stream, data, n):
    print(f"read {n} bytes from {stream.filepath}")

stream.hook("on_read", log_read)
stream.read()

Submodules

• duck.utils.fileio.test_fileio

Package Contents

Classes

AsyncFileIOStream	Asynchronous file streaming class.
FileIOStream	Synchronous file streaming class that mimics io.IOBase.

Functions

to_async_fileio_stream

Converts a FileIOStream to an AsyncFileIOStream if not already async.

Data

FILE_CACHE
VALID_EVENTS

API

class duck.utils.fileio.AsyncFileIOStream(*args, **kwargs)

Bases: duck.utils.fileio.FileIOStream

Asynchronous file streaming class.

Provides async-compatible methods for reading and writing files in a non-blocking way. Shares the same LRU cache and event-hook system as FileIOStream.

Writes update the cache directly with the written bytes, matching the synchronous behaviour. Async hooks registered on this stream are awaited inside the lock; sync hooks are called inline.

… admonition:: Notes

Compatible with async context managers (async with).

Initialization

Initialize the FileIOStream object.

Parameters:

• filepath – Path to the file to be streamed.

• chunk_size – Maximum number of bytes to read/write at once. Defaults to 2 MB.

• open_now – Whether to open the file immediately. Defaults to False.

• mode – File open mode. Defaults to 'rb'.

• disable_path_traversal – Whether to remove .. in paths to avoid path traversal. Defaults to True.

async __aenter__()

async __aexit__(exc_type, exc, tb)

async async_open() → None

Asynchronously open the file.

async async_read_chunked(size: int) → bytes

Read up to size bytes from the file in chunk_size pieces.

Parameters:

size – Number of bytes requested. A negative value reads all content.

Returns:

Up to size bytes; fewer only if the file ends first.

async async_write_chunked(data: bytes) → int

Write data to the file in chunk_size pieces.

Parameters:

data – Data to write.

Returns:

Number of bytes written; less than len(data) if the file stops accepting data part-way.

Raises:

OSError – If the very first chunk fails. Later failures return the partial count instead, so position and cache stay accurate.

async close() → None

Asynchronously close the file.

async fire_hooks_async(hooks: list[Callable], data: bytes) → None

Fires all hooks, awaiting async ones and calling sync ones inline.

Unlike the base :meth:fire_hooks, this variant is itself a coroutine so it can be awaited inside the async lock without scheduling fire-and-forget tasks.

Parameters:

• hooks – The list of callables to fire.

• data – The bytes that triggered this event.

async read(size: int = -1) → bytes

Asynchronously read from the file.

Results are served from the LRU cache when the file has not been modified since the last read. On a cache miss the file is read in a thread, in chunk_size pieces, until size bytes are collected or the file ends, and the result is stored in the cache.

Fires all on_read hooks after a successful read.

Parameters:

size – Max bytes to read. A negative value reads full content.

Returns:

Data read from file. Shorter than size only at end of file.

async write(data: bytes) → int

Asynchronously write data to the file.

The data is written in chunk_size pieces until all of it is written, then flushed to disk. Every cached entry overlapping the written region is patched with the post-write mtime, so the next read of that region is served from cache without a disk round-trip.

Fires all on_write hooks after a successful write.

Parameters:

data – Bytes to write.

Returns:

Number of bytes written.

duck.utils.fileio.FILE_CACHE: duck.utils.caching.InMemoryCache

‘InMemoryCache(…)’

class duck.utils.fileio.FileIOStream(filepath: str, chunk_size: int = 2 * 1024 * 1024, open_now: bool = False, mode: str = 'rb', disable_path_traversal: bool = True)

Bases: io.IOBase

Synchronous file streaming class that mimics io.IOBase.

Provides chunked reading and writing with a shared LRU cache and a simple event-hook system for on_read and on_write.

Read results are served from cache when the file is unchanged since the last read. Writes update the cache directly with the written bytes so the next full read is served from cache without an extra disk round-trip.

Hooks are registered with :meth:hook and fired after every read or write.

Initialization

Initialize the FileIOStream object.

Parameters:

• filepath – Path to the file to be streamed.

• chunk_size – Maximum number of bytes to read/write at once. Defaults to 2 MB.

• open_now – Whether to open the file immediately. Defaults to False.

• mode – File open mode. Defaults to 'rb'.

• disable_path_traversal – Whether to remove .. in paths to avoid path traversal. Defaults to True.

__del__() → None

Ensure the file is closed on delete else it raises a RuntimeError.

Always calls the synchronous base close directly — __del__ can never be a coroutine, so we must not dispatch to the async override on AsyncFileIOStream.

__slots__

None

_accumulate_read_bytes(data: bytes) → None

Appends newly read bytes to the running _total_read_bytes buffer.

Parameters:

data – The bytes returned from the most recent read operation.

_get_stat_metadata() → Tuple[int, int]

Retrieves the file’s size and modification time in a single stat call.

Combines what would otherwise be two separate os.stat() calls (one for etag, one for last_modified) into one, so callers that need both values only pay the stat cost once per access.

Returns:

A (st_size, st_mtime_ns) tuple for the file, or (0, 0) if the file does not exist.

cache_get(size: int) → Optional[bytes]

Returns cached bytes for the current position and size, or None.

Each (filepath, pos, size) triple has its own independent cache slot, so reads from different offsets are served correctly without interfering with each other. Stale entries (mtime mismatch) are evicted on access.

Parameters:

size – The read size passed to read(), or -1 for a full read.

Returns:

Cached bytes if the entry exists and is fresh, else None.

cache_patch_on_write(write_pos: int, data: bytes) → None

Surgically updates every cached entry that overlaps the written region.

Rather than flushing all cached entries or re-reading the whole file, this method iterates over existing cache keys for this filepath and patches any entry whose byte range overlaps [write_pos, write_pos + len(data)). Entries that do not overlap are left untouched — they remain valid because their byte ranges were not affected by the write.

A new entry is always written for (write_pos, len(data)) so the exact slice just written is immediately cache-warm.

Entries that overlap but cannot be fully reconstructed from the in-memory write (e.g. partial overlaps at the boundary of a cached chunk) are evicted rather than storing incorrect data.

Parameters:

• write_pos – The file offset at which the write started.

• data – The bytes that were just written.

cache_set(pos: int, size: int, data: bytes) → None

Stores a read result in the cache keyed by position and size.

Records the mtime on the instance so is_modified can compare against it later without an extra cache lookup.

Parameters:

• pos – The file offset at which the read started.

• size – The read size used to build the cache key.

• data – The bytes to cache.

close() → None

Close the file.

current_mtime() → float

Returns the file’s current modification time from the OS (as nanoseconds).

Returns:

The st_mtime_ns value for this stream’s filepath, or 0.0 if the file does not exist.

property etag: str

Returns a stable ETag derived from the file’s size and mtime.

Built from st_size and st_mtime_ns (nanosecond precision is preferred over st_mtime to avoid false cache hits on rapid successive writes), formatted as "size-mtime_ns". Since the value is derived from the whole file’s metadata rather than any requested slice, it is identical for every byte range served from this file.

Returns:

The ETag string, e.g. '"12345-1699999999000000000"'.

property filepath: str

Returns the filepath for the stream.

fire_hooks(hooks: list[Callable], data: bytes) → None

Fires all hooks in the given list with (self, data, len(data)).

Sync hooks are called inline. Async hooks are scheduled as fire-and-forget tasks on the running loop, or run in a new loop when none is active.

Parameters:

• hooks – The list of callables to fire.

• data – The bytes that triggered this event.

get_pos() → int

Get the stream pos.

hook(event: str, fn: Callable) → None

Registers a hook function for the given event.

The hook is called after every matching operation with the signature::

fn(stream, data, byte_count)

where stream is this FileIOStream, data is the bytes that were read or written, and byte_count is len(data).

Both plain callables and async coroutine functions are accepted. Async hooks on a synchronous stream are scheduled fire-and-forget on the running event loop when one is available.

Parameters:

• event – One of "on_read" or "on_write".

• fn – The callable to register.

Raises:

ValueError – When event is not a recognised event name.

increment_pos(pos: int)

Increment the fileio pos - this does not seek to the pos at all.

property is_modified: bool

Whether the file has been modified since this stream last read it.

Returns False when no read has occurred yet — there is no baseline mtime to compare against.

Returns:

True if the file’s current mtime differs from the mtime recorded during the last read, False otherwise.

is_open() → bool

Check if the file is currently open.

property last_modified: str

Returns the file’s last-modified time as an HTTP-date string.

Formatted with email.utils.formatdate(usegmt=True) to produce an RFC 7231 compliant Last-Modified header value, derived from st_mtime_ns.

Returns:

The formatted Last-Modified header value.

make_cache_key(pos: int, size: int) → str

Builds the cache key for a read starting at pos of length size.

The key encodes the filepath, position, and size so that reads from different offsets occupy independent cache slots.

Parameters:

• pos – The file offset at which the read starts.

• size – Number of bytes requested, or -1 for a full read.

Returns:

pos:size”`.

Return type:

A string cache key of the form `”filepath

open() → None

Open the file using the provided mode.

raise_if_in_async_context(message: str) → None

Raise an error if used inside an async context.

read(size: int = -1) → bytes

Synchronously read data from the file.

Results are served from the LRU cache when the file has not been modified since the last read. On a cache miss the file is read in chunk_size pieces until size bytes are collected or the file ends, and the result is stored in the cache for future calls.

Fires all on_read hooks after a successful read.

Parameters:

size – Number of bytes to read. A negative value reads all content.

Returns:

File data as bytes. Shorter than size only at end of file.

read_chunked(size: int) → bytes

Read up to size bytes from the file in chunk_size pieces.

Parameters:

size – Number of bytes requested. A negative value reads all content.

Returns:

Up to size bytes; fewer only if the file ends first.

seek(offset: int, whence: int = os.SEEK_SET) → None

Move the file pointer to a new location.

tell() → int

Get the current position in the file.

update_pos(pos: int)

Update the fileio pos - this does not seek to the pos at all.

write(data: bytes) → int

Synchronously write data to the file.

The data is written in chunk_size pieces until all of it is written, then flushed to disk immediately. Every cached entry overlapping the written region is patched with the post-write mtime, so the next read of that region is served from cache without a disk round-trip.

Fires all on_write hooks after a successful write.

Parameters:

data – Data to write.

Returns:

Number of bytes written.

write_chunked(data: bytes) → int

Write data to the file in chunk_size pieces.

Parameters:

data – Data to write.

Returns:

Number of bytes written; less than len(data) if the file stops accepting data part-way.

Raises:

OSError – If the very first chunk fails. Later failures return the partial count instead, so position and cache stay accurate.

exception duck.utils.fileio.PathTraversalWarning

Bases: UserWarning

Warning for path traversal vulnerability.

Initialization

Initialize self. See help(type(self)) for accurate signature.

duck.utils.fileio.VALID_EVENTS

‘frozenset(…)’

duck.utils.fileio.to_async_fileio_stream(fileio_stream: FileIOStream) → AsyncFileIOStream

Converts a FileIOStream to an AsyncFileIOStream if not already async.

Parameters:

fileio_stream – The synchronous stream to convert.

Returns:

An AsyncFileIOStream wrapping the same underlying file descriptor.