dulwich.cli.AutoFlushTextIOWrapper

class documentation

class AutoFlushTextIOWrapper:

Constructor: AutoFlushTextIOWrapper(stream)

View In Hierarchy

Wrapper that automatically flushes a TextIO stream based on configuration.

This wrapper can be configured to flush after each write operation, which is useful for real-time output monitoring in CI/CD systems.

Class Method env Create wrapper respecting the GIT_FLUSH environment variable.
Method __enter__ Support context manager protocol.
Method __exit__ Support context manager protocol.
Method __getattr__ Delegate all other attributes to the underlying stream.
Method __init__ Initialize the wrapper.
Method flush Flush the underlying stream.
Method write Write data to the stream and optionally flush.
Method writelines Write multiple lines to the stream and optionally flush.
Instance Variable _stream Undocumented
def env(cls, stream: TextIO, env: Mapping[str, str] | None = None) -> AutoFlushTextIOWrapper | TextIO:

Create wrapper respecting the GIT_FLUSH environment variable.

Respects the GIT_FLUSH environment variable: - GIT_FLUSH=1: Always flush after each write - GIT_FLUSH=0: Never auto-flush (use buffered I/O) - Not set: Auto-detect based on whether output is redirected

Parameters
stream:TextIOThe stream to wrap
env:Mapping[str, str] | NoneEnvironment variables dict (defaults to os.environ)
Returns
AutoFlushTextIOWrapper | TextIOAutoFlushTextIOWrapper instance configured based on GIT_FLUSH
def __enter__(self) -> AutoFlushTextIOWrapper:

Support context manager protocol.

def __exit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None):

Support context manager protocol.

def __getattr__(self, name: str) -> object:

Delegate all other attributes to the underlying stream.

def __init__(self, stream: TextIO):

Initialize the wrapper.

Parameters
stream:TextIOThe stream to wrap
def flush(self):

Flush the underlying stream.

def write(self, data: str) -> int:

Write data to the stream and optionally flush.

Parameters
data:strData to write
Returns
intNumber of characters written
def writelines(self, lines: Iterable[str]):

Write multiple lines to the stream and optionally flush.

Parameters
lines:Iterable[str]Lines to write
_stream =

Undocumented