Cleo allows you to create beautiful and testable command-line interfaces.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
The I/O system provides comprehensive input and output management for CLI applications. It handles input from various sources (command line arguments, strings, user interaction), manages output to different destinations (stdout, stderr, files, memory buffers), and supports verbosity levels and formatting.
The main IO class coordinates input, output, and error output streams with verbosity control.
class IO:
def __init__(self, input: Input, output: Output, error_output: Output) -> None:
"""
Create an IO coordinator.
Args:
input (Input): Input stream for reading user input
output (Output): Primary output stream
error_output (Output): Error output stream
"""
def read(self, length: int, default: str = "") -> str:
"""
Read input from the input stream.
Args:
length (int): Maximum length to read
default (str): Default value if no input
Returns:
str: Input text
"""
def write(self, messages: str | Iterable[str], new_line: bool = False) -> None:
"""
Write messages to output.
Args:
messages (str | Iterable[str]): Messages to write
new_line (bool): Whether to add newline after each message
"""
def write_line(self, messages: str | Iterable[str]) -> None:
"""
Write messages to output with newlines.
Args:
messages (str | Iterable[str]): Messages to write
"""
def write_error(self, messages: str | Iterable[str], new_line: bool = False) -> None:
"""
Write messages to error output.
Args:
messages (str | Iterable[str]): Error messages to write
new_line (bool): Whether to add newline
"""
def write_error_line(self, messages: str | Iterable[str]) -> None:
"""
Write messages to error output with newlines.
Args:
messages (str | Iterable[str]): Error messages to write
"""
def is_interactive(self) -> bool:
"""
Check if input is interactive (TTY).
Returns:
bool: True if interactive
"""
def is_decorated(self) -> bool:
"""
Check if output supports decoration (colors/formatting).
Returns:
bool: True if decorated output is supported
"""
def set_verbosity(self, verbosity: Verbosity) -> None:
"""
Set the verbosity level.
Args:
verbosity (Verbosity): Verbosity level to set
"""
@property
def verbosity(self) -> Verbosity:
"""Get the current verbosity level."""
@property
def input(self) -> Input:
"""Get the input stream."""
@property
def output(self) -> Output:
"""Get the output stream."""
@property
def error_output(self) -> Output:
"""Get the error output stream."""Alternative IO implementations for different use cases.
class BufferedIO(IO):
"""IO implementation that buffers output in memory for testing."""
def fetch_output(self) -> str:
"""
Fetch the buffered output content.
Returns:
str: All output written since last fetch
"""
def fetch_error(self) -> str:
"""
Fetch the buffered error output content.
Returns:
str: All error output written since last fetch
"""
class NullIO(IO):
"""IO implementation that discards all output."""
passAbstract base class for all input sources.
class Input:
def __init__(self, definition: Definition | None = None) -> None: ...
def bind(self, definition: Definition) -> None:
"""Bind input to a command definition."""
def validate(self) -> None:
"""Validate input against bound definition."""
def get_arguments(self) -> dict[str, Any]:
"""Get all arguments as dictionary."""
def get_argument(self, name: str) -> Any:
"""Get argument value by name."""
def set_argument(self, name: str, value: Any) -> None:
"""Set argument value."""
def has_argument(self, name: str) -> bool:
"""Check if argument exists."""
def get_options(self) -> dict[str, Any]:
"""Get all options as dictionary."""
def get_option(self, name: str) -> Any:
"""Get option value by name."""
def set_option(self, name: str, value: Any) -> None:
"""Set option value."""
def has_option(self, name: str) -> bool:
"""Check if option exists."""
def is_interactive(self) -> bool:
"""Check if input is interactive."""
@property
def definition(self) -> Definition:
"""Get the input definition."""Concrete input classes for different input sources.
class ArgvInput(Input):
"""Input from command line arguments (sys.argv)."""
def __init__(self, argv: list[str] | None = None, definition: Definition | None = None) -> None:
"""
Create input from command line arguments.
Args:
argv (list[str] | None): Argument list, defaults to sys.argv
definition (Definition | None): Input definition
"""
class StringInput(Input):
"""Input from a string for testing."""
def __init__(self, input_str: str, definition: Definition | None = None) -> None:
"""
Create input from a string.
Args:
input_str (str): Input string to parse
definition (Definition | None): Input definition
"""Abstract base class for all output destinations.
class Output:
def __init__(self, verbosity: Verbosity = Verbosity.NORMAL, decorated: bool | None = None,
formatter: Formatter | None = None) -> None: ...
def write(self, messages: str | Iterable[str], new_line: bool = False,
type: Type = Type.NORMAL) -> None:
"""
Write messages to output.
Args:
messages (str | Iterable[str]): Messages to write
new_line (bool): Whether to add newlines
type (Type): Output type (NORMAL, RAW, PLAIN)
"""
def write_line(self, messages: str | Iterable[str], type: Type = Type.NORMAL) -> None:
"""Write messages with newlines."""
def set_verbosity(self, level: Verbosity) -> None:
"""Set verbosity level."""
def get_verbosity(self) -> Verbosity:
"""Get current verbosity level."""
def is_quiet(self) -> bool:
"""Check if output is in quiet mode."""
def is_verbose(self) -> bool:
"""Check if output is in verbose mode."""
def is_very_verbose(self) -> bool:
"""Check if output is in very verbose mode."""
def is_debug(self) -> bool:
"""Check if output is in debug mode."""
def set_decorated(self, decorated: bool) -> None:
"""Enable/disable output decoration."""
def is_decorated(self) -> bool:
"""Check if output is decorated."""
def set_formatter(self, formatter: Formatter) -> None:
"""Set the output formatter."""
def get_formatter(self) -> Formatter:
"""Get the output formatter."""Concrete output classes for different output destinations.
class StreamOutput(Output):
"""Output to a stream (stdout, stderr, file)."""
def __init__(self, stream: TextIO, verbosity: Verbosity = Verbosity.NORMAL,
decorated: bool | None = None, formatter: Formatter | None = None) -> None:
"""
Create output to a stream.
Args:
stream (TextIO): Output stream
verbosity (Verbosity): Verbosity level
decorated (bool | None): Whether to use decoration
formatter (Formatter | None): Output formatter
"""
class BufferedOutput(Output):
"""Output buffered in memory."""
def fetch(self) -> str:
"""
Fetch and clear the buffer content.
Returns:
str: Buffered output content
"""
def get_content(self) -> str:
"""
Get buffer content without clearing.
Returns:
str: Current buffer content
"""
def clear(self) -> None:
"""Clear the buffer."""
class NullOutput(Output):
"""Output that discards all content."""
pass
class SectionOutput(Output):
"""Output that can be updated in sections."""
def clear(self, lines: int | None = None) -> None:
"""
Clear lines from output.
Args:
lines (int | None): Number of lines to clear
"""
def overwrite(self, messages: str | Iterable[str]) -> None:
"""
Overwrite the current section content.
Args:
messages (str | Iterable[str]): New content
"""Enumeration of verbosity levels for controlling output detail.
class Verbosity(Enum):
QUIET = 16 # Suppress all output except errors
NORMAL = 32 # Normal output level
VERBOSE = 64 # Increased verbosity (-v)
VERY_VERBOSE = 128 # High verbosity (-vv)
DEBUG = 256 # Debug information (-vvv)Enumeration of output formatting types.
class Type(Enum):
NORMAL = 1 # Normal formatted output
RAW = 2 # Raw output without formatting
PLAIN = 4 # Plain output without colorsfrom cleo.io.io import IO
from cleo.io.inputs.argv_input import ArgvInput
from cleo.io.outputs.stream_output import StreamOutput
import sys
# Create IO with standard streams
input = ArgvInput()
output = StreamOutput(sys.stdout)
error_output = StreamOutput(sys.stderr)
io = IO(input, output, error_output)
# Write output
io.write_line("Processing data...")
io.write_error_line("Warning: deprecated option used")from cleo.io.buffered_io import BufferedIO
from cleo.io.inputs.string_input import StringInput
# Create buffered IO for testing
input = StringInput("test-file --verbose")
io = BufferedIO(input)
# Run command with buffered IO
command.run(io)
# Check output
output = io.fetch_output()
errors = io.fetch_error()
assert "Success" in outputfrom cleo.io.outputs.output import Verbosity
# Set verbosity level
io.set_verbosity(Verbosity.VERBOSE)
# Check verbosity in commands
if io.output.is_verbose():
io.write_line("Detailed processing information...")
if io.output.is_debug():
io.write_line(f"Debug: processing {len(items)} items")
# Write with specific verbosity requirements
io.write_line("Always shown", verbosity=Verbosity.NORMAL)
io.write_line("Only in verbose mode", verbosity=Verbosity.VERBOSE)from cleo.io.outputs.section_output import SectionOutput
# Create section output for live updates
section = output.section()
# Update progress
for i in range(100):
section.overwrite(f"Progress: {i}%")
time.sleep(0.1)
section.overwrite("Complete!")Install with Tessl CLI
npx tessl i tessl/pypi-cleo