tron.serialize.filehandler module

Tools for managing and properly closing file handles.

class tron.serialize.filehandler.FileHandleManager(max_idle_time=60)

Bases: object

Creates FileHandleWrappers, closes handles when they have been inactive for a period of time, and transparently re-open the next time they are needed. All files are opened in append mode.

This class is singleton. An already configured instance can be retrieving by using get_instance() (and will be created if None), max_idle_time can be set by calling the classmethod set_max_idle_time()

cleanup(time_func=<built-in function time>)

Close any file handles that have been idle for longer than max_idle_time. time_func is primary used for testing.

classmethod get_instance()

open(filename)

Retrieve a file handle from the cache based on name. Returns a FileHandleWrapper. If the handle is not in the cache, create a new instance.

remove(fh_wrapper)

Remove the fh_wrapper from the cache and access_order.

classmethod reset()

Empty the cache and reset the instance to it’s original state.

classmethod set_max_idle_time(max_idle_time)

update(fh_wrapper)

Remove and re-add the file handle to the cache so that it’s keys are still ordered by last access. Calls cleanup() to remove any file handles that have been idle for too long.

class tron.serialize.filehandler.FileHandleWrapper(manager, name)

Bases: object

Acts as a proxy to file handles. Wrap a file handle and stores access time and metadata. These objects should only be created by FileHandleManager. Do not instantiate them on their own.

close()

close_wrapped()

Close only the underlying file handle.

last_accessed

manager

name

write(content)

Write content to the fh. Re-open if necessary.

class tron.serialize.filehandler.NullFileHandle

Bases: object

A No-Op object that supports a File interface.

classmethod close()

closed = True

classmethod write(_)

class tron.serialize.filehandler.OutputPath(base='.', *path_parts)

Bases: object

A list like object used to construct a file path for output. The file path is constructed by joining the base path with any additional path elements.

append(part)

base

clone(*parts)

Return a new OutputPath object which has a base of the str value of this object.

delete()

Remove the directory and its contents.

parts

class tron.serialize.filehandler.OutputStreamSerializer(base_path)

Bases: object

Manage writing to and reading from files in a directory hierarchy.

full_path(filename)

open(filename)

Return a FileHandleManager for the output path.

tail(filename: str, num_lines: int | None = None) → list[str]

Tail a file using tail.