Rotary Logger  1.0.2
The middleware rotary logger
Loading...
Searching...
No Matches
Public Member Functions | Data Fields | Protected Member Functions | Protected Attributes
rotary_logger.tee_stream.TeeStream Class Reference
Collaboration diagram for rotary_logger.tee_stream.TeeStream:
Collaboration graph

Public Member Functions

 __init__ (self, Union[str, Path, FileInstance] root, TextIO original_stream, *, Optional[int] max_size_mb=None, Optional[int] flush_size=None, CONST.StdMode mode=CONST.StdMode.STDUNKNOWN, CONST.ErrorMode error_mode=CONST.ErrorMode.WARN_NO_PIPE, Optional[str] encoding=None, bool log_to_file=True, bool log_function_calls=False)
 
 __del__ (self)
 
None write (self, str message)
 
None writelines (self, List[str] lines)
 
str read (self, int size=-1)
 
str readline (self, int size=-1)
 
list[str] readlines (self, int hint=-1)
 
None flush (self)
 
BinaryIO buffer (self)
 
bool closed (self)
 
Optional[str] errors (self)
 
str encoding (self)
 
int line_buffering (self)
 
str mode (self)
 
Union[str, Any] name (self)
 
Any newlines (self)
 
None close (self)
 
int fileno (self)
 
bool isatty (self)
 
bool readable (self)
 
int seek (self, int offset, int whence=0)
 
bool seekable (self)
 
int tell (self)
 
int truncate (self, Optional[int] size=None)
 
bool writable (self)
 
TextIO __enter__ (self)
 
Optional[bool] __exit__ (self, exc_type, exc_val, exc_tb)
 
 __iter__ (self)
 
 __next__ (self)
 
 detach (self)
 
 __getattr__ (self, str name)
 

Data Fields

 file_instance = FileInstance(Path(root))
 
CONST.StdMode stream_mode = mode
 
TextIO original_stream = original_stream
 
CONST.ErrorMode error_mode = error_mode
 
AttributeError stream_not_present
 
 function_calls = log_function_calls
 
Rogger rogger = RI
 
str function_calls
 

Protected Member Functions

str _get_correct_prefix (self, CONST.PrefixFunctionCall function_call=CONST.PrefixFunctionCall.EMPTY)
 
None _write_to_log (self, Union[str, List[str]] data, CONST.PrefixFunctionCall function_call)
 
TextIO _get_stream_if_present (self)
 

Protected Attributes

RLock _file_lock = RLock()
 

Detailed Description

Mirror a TextIO stream to disk while preserving normal output.

This class is intentionally lightweight: it captures short-lived
references under a tiny lock and then performs I/O without holding
that lock. Terminal writes are performed on the caller thread and
are wrapped to avoid raising unexpected errors back into the
application; disk writes are delegated to a `FileInstance` which
buffers and handles rotation.

Definition at line 50 of file tee_stream.py.

Constructor & Destructor Documentation

◆ __init__()

rotary_logger.tee_stream.TeeStream.__init__ ( self,
Union[str, Path, FileInstance] root,
TextIO original_stream,
* ,
Optional[int] max_size_mb = None,
Optional[int] flush_size = None,
CONST.StdMode mode = CONST.StdMode.STDUNKNOWN,
CONST.ErrorMode error_mode = CONST.ErrorMode.WARN_NO_PIPE,
Optional[str] encoding = None,
bool log_to_file = True,
bool log_function_calls = False ) 
Initialise a new TeeStream.

Mirrors original_stream to disk via a FileInstance while forwarding
every write transparently to the caller.

Arguments:
    root (Union[str, Path, FileInstance]): Path, string, or FileInstance describing the log destination.
    original_stream (TextIO): The TextIO stream to mirror (usually sys.stdout).

Keyword Arguments:
    max_size_mb (Optional[int]): Optional maximum log-file size in MB; forwarded to FileInstance. Default: None
    flush_size (Optional[int]): Optional buffer-flush threshold in bytes; forwarded to FileInstance. Default: None
    mode (CONST.StdMode): Which standard stream this instance wraps. Default: CONST.StdMode.STDUNKNOWN
    error_mode (CONST.ErrorMode): Broken-pipe handling policy. Default: CONST.ErrorMode.WARN_NO_PIPE
    encoding (Optional[str]): Optional file-encoding override; forwarded to FileInstance. Default: None
    log_to_file (bool): Whether disk logging is enabled on construction. Default: True

Raises:
    ValueError: If root is not a str, Path, or FileInstance.

Definition at line 61 of file tee_stream.py.

◆ __del__()

rotary_logger.tee_stream.TeeStream.__del__ ( self) 
Best-effort cleanup on object deletion.

Attempts to flush buffered data but never raises. Note that __del__
may not be called at interpreter shutdown; callers that need
deterministic flushing should invoke flush() explicitly before
releasing the object. OSError and ValueError are intentionally
swallowed so that interpreter-shutdown teardown cannot trigger
unexpected tracebacks.

Definition at line 129 of file tee_stream.py.

Member Function Documentation

◆ __enter__()

TextIO rotary_logger.tee_stream.TeeStream.__enter__ ( self) 
Enter the runtime context for the original stream.

Delegates to the wrapped stream's __enter__() so that TeeStream can be
used as a context manager wherever a plain TextIO is expected.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The original stream as returned by its own __enter__().

Definition at line 797 of file tee_stream.py.

◆ __exit__()

Optional[bool] rotary_logger.tee_stream.TeeStream.__exit__ ( self,
exc_type,
exc_val,
exc_tb ) 
Exit the runtime context and delegate to the original stream.

Delegates to the wrapped stream's __exit__() so that TeeStream can be
used as a context manager wherever a plain TextIO is expected.

Arguments:
    exc_type (type): The exception type, or None if no exception.
    exc_val (BaseException): The exception instance, or None.
    exc_tb (traceback): The traceback, or None.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The return value of the original stream's __exit__(); True suppresses
    the exception, False or None propagates it.

Definition at line 812 of file tee_stream.py.

◆ __getattr__()

rotary_logger.tee_stream.TeeStream.__getattr__ ( self,
str name ) 
Delegate attribute access to the wrapped stream for missing attrs.

This helps the TeeStream more closely mimic the wrapped TextIO
object without having to manually proxy every attribute.

Definition at line 887 of file tee_stream.py.

◆ __iter__()

rotary_logger.tee_stream.TeeStream.__iter__ ( self) 
Return an iterator over the stream (line-by-line).

This makes `for line in tee_stream:` work like the underlying
TextIO object. Iteration yields lines as produced by `readline()`.

Definition at line 837 of file tee_stream.py.

◆ __next__()

rotary_logger.tee_stream.TeeStream.__next__ ( self) 
Return the next line from the wrapped stream, logging it.

Prefer delegating to the underlying stream's `__next__` if it
exists (and then log the returned line). Otherwise fall back to
`readline()` which already takes care of logging. Raise
`StopIteration` when the stream is exhausted.

Definition at line 845 of file tee_stream.py.

◆ _get_correct_prefix()

str rotary_logger.tee_stream.TeeStream._get_correct_prefix ( self,
CONST.PrefixFunctionCall function_call = CONST.PrefixFunctionCall.EMPTY )
protected 
Return the correct prefix string for the configured StdMode.

The returned string already contains a trailing space when non-empty so
that callers can concatenate it with the message directly. Returns an
empty string when logging to file is disabled, when no FileInstance is
set, or when the Prefix configuration has no flags enabled.

Returns:
    The prefix string (with trailing space) matching the active StdMode,
    or an empty string.

Definition at line 147 of file tee_stream.py.

◆ _get_stream_if_present()

TextIO rotary_logger.tee_stream.TeeStream._get_stream_if_present ( self)
protected 
Return the underlying stream, raising if it has been cleared.

Used as a guard by every delegating method so that operations on an
uninitialised or already-destroyed TeeStream fail predictably rather
than with an obscure AttributeError.

Raises:
    AttributeError: If original_stream is None.

Returns:
    The original TextIO stream passed at construction time.

Definition at line 264 of file tee_stream.py.

◆ _write_to_log()

None rotary_logger.tee_stream.TeeStream._write_to_log ( self,
Union[str, List[str]] data,
CONST.PrefixFunctionCall function_call )
protected 
Write data to the log file if file logging is enabled.

Shared helper used by write(), writelines(), read(), readline() and
readlines() to avoid duplicating the snapshot-check-prefix-write
sequence in every method.

Arguments:
    data (str): The string to append to the log file.
    function_call (CONST.PrefixFunctionCall): Context passed to
        _get_correct_prefix() to select the right prefix.

Definition at line 215 of file tee_stream.py.

◆ buffer()

BinaryIO rotary_logger.tee_stream.TeeStream.buffer ( self) 
Return the underlying binary buffer of the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The binary buffer exposed by the wrapped TextIO.

Definition at line 542 of file tee_stream.py.

◆ close()

None rotary_logger.tee_stream.TeeStream.close ( self) 
Flush pending data and close the original stream.

Calls flush() before delegating to the original stream's close()
method, ensuring buffered log data is written before the stream is
released.

Raises:
    AttributeError: If the original stream is not set.

Definition at line 657 of file tee_stream.py.

◆ closed()

bool rotary_logger.tee_stream.TeeStream.closed ( self) 
Return whether the original stream is closed.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    True if the wrapped stream has been closed, False otherwise.

Definition at line 555 of file tee_stream.py.

◆ detach()

rotary_logger.tee_stream.TeeStream.detach ( self) 
Detach and return the underlying binary buffer if supported.

Delegates to the wrapped stream's `detach()` when present. If the
wrapped stream does not support `detach()` an AttributeError is
raised to mirror standard behavior.

Definition at line 873 of file tee_stream.py.

◆ encoding()

str rotary_logger.tee_stream.TeeStream.encoding ( self) 
Return the character encoding of the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The encoding string (e.g. 'utf-8').

Definition at line 583 of file tee_stream.py.

◆ errors()

Optional[str] rotary_logger.tee_stream.TeeStream.errors ( self) 
Return the error-handling mode of the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The error-handling mode string (e.g. 'strict'), or None.

Definition at line 569 of file tee_stream.py.

◆ fileno()

int rotary_logger.tee_stream.TeeStream.fileno ( self) 
Return the underlying file descriptor of the original stream.

Raises:
    AttributeError: If the original stream is not set.
    io.UnsupportedOperation: If the stream has no file descriptor.

Returns:
    Integer file descriptor.

Definition at line 677 of file tee_stream.py.

◆ flush()

None rotary_logger.tee_stream.TeeStream.flush ( self) 
Best-effort flush of terminal and buffered file output.

Attempts to flush both the original stream and the associated
FileInstance. All OSError and ValueError exceptions are swallowed to
avoid crashing the caller. For stricter guarantees call
FileInstance.flush() directly. This method is also called by __del__
during object teardown.

Definition at line 454 of file tee_stream.py.

◆ isatty()

bool rotary_logger.tee_stream.TeeStream.isatty ( self) 
Return whether the original stream is connected to a TTY device.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    True if the stream is interactive (TTY), False otherwise.

Definition at line 691 of file tee_stream.py.

◆ line_buffering()

int rotary_logger.tee_stream.TeeStream.line_buffering ( self) 
Return whether line buffering is enabled on the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    Non-zero if line buffering is active, zero otherwise.

Definition at line 597 of file tee_stream.py.

◆ mode()

str rotary_logger.tee_stream.TeeStream.mode ( self) 
Return the file-mode string of the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The mode string (e.g. 'w', 'r').

Definition at line 614 of file tee_stream.py.

◆ name()

Union[str, Any] rotary_logger.tee_stream.TeeStream.name ( self) 
Return the name of the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The stream name; typically a file path string or an integer file
    descriptor for standard streams.

Definition at line 628 of file tee_stream.py.

◆ newlines()

Any rotary_logger.tee_stream.TeeStream.newlines ( self) 
Return the newline translation used by the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    A string, tuple of strings, or None as described by the standard
    io.TextIOBase.newlines attribute.

Definition at line 643 of file tee_stream.py.

◆ read()

str rotary_logger.tee_stream.TeeStream.read ( self,
int size = -1 ) 
Read and return up to size characters from the original stream.

Keyword Arguments:
    size (int): Maximum number of characters to read; -1 reads until EOF. Default: -1

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The characters read.

Definition at line 380 of file tee_stream.py.

◆ readable()

bool rotary_logger.tee_stream.TeeStream.readable ( self) 
Return whether the original stream supports reading.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    True if the stream can be read from, False otherwise.

Definition at line 704 of file tee_stream.py.

◆ readline()

str rotary_logger.tee_stream.TeeStream.readline ( self,
int size = -1 ) 
Read and return one line from the original stream.

Keyword Arguments:
    size (int): If non-negative, at most size characters are read; -1 reads until a newline or EOF. Default: -1

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The line read, including the trailing newline if present.

Definition at line 405 of file tee_stream.py.

◆ readlines()

list[str] rotary_logger.tee_stream.TeeStream.readlines ( self,
int hint = -1 ) 
Read and return a list of lines from the original stream.

Keyword Arguments:
    hint (int): If non-negative, approximately hint bytes or characters are read; -1 reads until EOF. Default: -1

Raises:
    AttributeError: If the original stream is not set.

Returns:
    List of lines read from the stream.

Definition at line 428 of file tee_stream.py.

◆ seek()

int rotary_logger.tee_stream.TeeStream.seek ( self,
int offset,
int whence = 0 ) 
Move the stream position to the given byte offset.

Arguments:
    offset (int): Number of bytes to move the position.

Keyword Arguments:
    whence (int): Reference point: 0 = start, 1 = current, 2 = end. Default: 0

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The new absolute stream position.

Definition at line 717 of file tee_stream.py.

◆ seekable()

bool rotary_logger.tee_stream.TeeStream.seekable ( self) 
Return whether the original stream supports random-access seeking.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    True if the stream supports seek() and tell(), False otherwise.

Definition at line 739 of file tee_stream.py.

◆ tell()

int rotary_logger.tee_stream.TeeStream.tell ( self) 
Return the current stream position of the original stream.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The current byte offset within the stream.

Definition at line 752 of file tee_stream.py.

◆ truncate()

int rotary_logger.tee_stream.TeeStream.truncate ( self,
Optional[int] size = None ) 
Truncate the original stream to at most size bytes.

Keyword Arguments:
    size (Optional[int]): Desired size in bytes; defaults to the current stream position. Default: None

Raises:
    AttributeError: If the original stream is not set.

Returns:
    The new file size in bytes.

Definition at line 765 of file tee_stream.py.

◆ writable()

bool rotary_logger.tee_stream.TeeStream.writable ( self) 
Return whether the original stream supports writing.

Raises:
    AttributeError: If the original stream is not set.

Returns:
    True if the stream can be written to, False otherwise.

Definition at line 784 of file tee_stream.py.

◆ write()

None rotary_logger.tee_stream.TeeStream.write ( self,
str message ) 
Write message to the original stream and buffer it to the log file.

Thread-safe: the FileInstance reference is captured under a lock before
any I/O is performed. The terminal write is carried out on the caller
thread with explicit BrokenPipeError / OSError handling; disk writes
are delegated to FileInstance.write().

Arguments:
    message (str): The string to write.

Definition at line 284 of file tee_stream.py.

◆ writelines()

None rotary_logger.tee_stream.TeeStream.writelines ( self,
List[str] lines ) 
Write a list of strings to the original stream and buffer them to the log file.

Thread-safe: behaves like write() but accepts a sequence of strings,
forwarding the entire sequence to the original stream via writelines()
and then writing each element individually to FileInstance with the
appropriate prefix.

Arguments:
    lines (List[str]): The sequence of strings to write.

Definition at line 332 of file tee_stream.py.

Field Documentation

◆ _file_lock

RLock rotary_logger.tee_stream.TeeStream._file_lock = RLock()
protected

Definition at line 95 of file tee_stream.py.

◆ error_mode

CONST.ErrorMode rotary_logger.tee_stream.TeeStream.error_mode = error_mode

Definition at line 113 of file tee_stream.py.

◆ file_instance

rotary_logger.tee_stream.TeeStream.file_instance = FileInstance(Path(root))

Definition at line 97 of file tee_stream.py.

◆ function_calls [1/2]

rotary_logger.tee_stream.TeeStream.function_calls = log_function_calls

Definition at line 117 of file tee_stream.py.

◆ function_calls [2/2]

str rotary_logger.tee_stream.TeeStream.function_calls
Initial value:
= self._get_correct_prefix(
function_call=CONST.PrefixFunctionCall.FLUSH
)

Definition at line 513 of file tee_stream.py.

◆ original_stream

TextIO rotary_logger.tee_stream.TeeStream.original_stream = original_stream

Definition at line 112 of file tee_stream.py.

◆ rogger

Rogger rotary_logger.tee_stream.TeeStream.rogger = RI

Definition at line 118 of file tee_stream.py.

◆ stream_mode

CONST.StdMode rotary_logger.tee_stream.TeeStream.stream_mode = mode

Definition at line 111 of file tee_stream.py.

◆ stream_not_present

AttributeError rotary_logger.tee_stream.TeeStream.stream_not_present
Initial value:
= AttributeError(
f"{CONST.MODULE_NAME} No stream available"
)

Definition at line 114 of file tee_stream.py.

The documentation for this class was generated from the following file: