dulwich.cli.AutoFlushBinaryIOWrapper

class documentation

class AutoFlushBinaryIOWrapper:

Constructor: AutoFlushBinaryIOWrapper(stream)

Wrapper that automatically flushes a BinaryIO 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

@classmethod

def env(cls, stream: BinaryIO, env: Mapping[str, str] | None = None) -> AutoFlushBinaryIOWrapper | BinaryIO: ¶

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:`BinaryIO`	The stream to wrap
env:`Mapping[str, str] \| None`	Environment variables dict (defaults to os.environ)

Returns
`AutoFlushBinaryIOWrapper \| BinaryIO`	AutoFlushBinaryIOWrapper instance configured based on GIT_FLUSH

def __enter__(self) -> AutoFlushBinaryIOWrapper: ¶

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: BinaryIO): ¶

Initialize the wrapper.

Parameters
stream:`BinaryIO`	The stream to wrap

def flush(self): ¶

Flush the underlying stream.

def write(self, data: Buffer) -> int: ¶

Write data to the stream and optionally flush.

Parameters
data:`Buffer`	Data to write

Returns
`int`	Number of bytes written

def writelines(self, lines: Iterable[Buffer]): ¶

Write multiple lines to the stream and optionally flush.

Parameters
lines:`Iterable[Buffer]`	Lines to write

_stream = ¶

Undocumented