Console

provide.foundation.console

TODO: Add module docstring.

Functions

apin async

apin(prompt: str = '', **kwargs: Any) -> str | Any

Async input from stdin with optional prompt.

Parameters:

Name	Type	Description	Default
prompt	str	Prompt to display before input	''
**kwargs	Any	Same as pin()	{}

Returns:

Type	Description
str | Any	User input as string or converted type

Examples:

name = await apin("Enter name: ") age = await apin("Age: ", type=int)

Note: This runs the blocking input in a thread pool to avoid blocking the event loop.

Source code in provide/foundation/console/input.py

async def apin(prompt: str = "", **kwargs: Any) -> str | Any:
    """Async input from stdin with optional prompt.

    Args:
        prompt: Prompt to display before input
        **kwargs: Same as pin()

    Returns:
        User input as string or converted type

    Examples:
        name = await apin("Enter name: ")
        age = await apin("Age: ", type=int)

    Note: This runs the blocking input in a thread pool to avoid blocking the event loop.

    """
    import functools

    loop = asyncio.get_event_loop()
    func = functools.partial(pin, prompt, **kwargs)
    return await loop.run_in_executor(None, func)

apin_lines async

apin_lines(count: int | None = None) -> list[str]

Async read multiple lines from stdin.

Parameters:

Name	Type	Description	Default
count	int | None	Number of lines to read (None for all until EOF)	None

Returns:

Type	Description
list[str]	List of input lines

Examples:

lines = await apin_lines(3) # Read exactly 3 lines all_lines = await apin_lines() # Read until EOF

Source code in provide/foundation/console/input.py

async def apin_lines(count: int | None = None) -> list[str]:
    """Async read multiple lines from stdin.

    Args:
        count: Number of lines to read (None for all until EOF)

    Returns:
        List of input lines

    Examples:
        lines = await apin_lines(3)  # Read exactly 3 lines
        all_lines = await apin_lines()  # Read until EOF

    """
    lines = []
    i = 0
    async for line in apin_stream():
        lines.append(line)
        i += 1
        if count is not None and i >= count:
            break
    return lines

apin_stream async

apin_stream() -> AsyncIterator[str]

Async stream input line by line from stdin.

Yields:

Type	Description
AsyncIterator[str]	Lines from stdin (without trailing newline)

Examples:

async for line in apin_stream(): await process(line)

This provides non-blocking line-by-line input streaming.

Source code in provide/foundation/console/input.py

async def apin_stream() -> AsyncIterator[str]:
    """Async stream input line by line from stdin.

    Yields:
        Lines from stdin (without trailing newline)

    Examples:
        async for line in apin_stream():
            await process(line)

    This provides non-blocking line-by-line input streaming.

    """
    ctx = _get_context()

    if _should_use_json(ctx):
        # In JSON mode, read all input and yield parsed lines
        loop = asyncio.get_event_loop()

        def read_json() -> list[str]:
            try:
                stdin_content = sys.stdin.read()
                data = json_loads(stdin_content)
                if isinstance(data, list):
                    return [json_dumps(item) if not isinstance(item, str) else item for item in data]
                return [json_dumps(data)]
            except ValidationError:
                # Fall back to line-by-line reading - content already read
                return [line.rstrip("\n\r") for line in stdin_content.splitlines() if line]

        lines = await loop.run_in_executor(None, read_json)
        for line in lines:
            yield line
    else:
        # Regular mode - async line streaming
        log.debug("📥 Starting async input stream")
        line_count = 0

        # Create async reader for stdin
        loop = asyncio.get_event_loop()
        reader = asyncio.StreamReader()
        protocol = asyncio.StreamReaderProtocol(reader)

        await loop.connect_read_pipe(lambda: protocol, sys.stdin)

        try:
            while True:
                try:
                    line_bytes = await reader.readline()
                    if not line_bytes:
                        break

                    line = line_bytes.decode("utf-8").rstrip("\n\r")
                    line_count += 1
                    log.trace("📥 Async stream line", line_num=line_count, length=len(line))
                    yield line

                except asyncio.CancelledError:
                    log.debug("📥 Async stream cancelled", lines=line_count)
                    break
                except Exception as e:
                    log.error("📥 Async stream error", error=str(e), lines=line_count)
                    break
        finally:
            log.debug("📥 Async input stream ended", lines=line_count)

perr

perr(message: Any, **kwargs: Any) -> None

Output message to stderr.

Parameters:

Name	Type	Description	Default
message	Any	Content to output (any type - will be stringified or JSON-encoded)	required
**kwargs	Any	Optional formatting arguments: color: Color name (red, green, yellow, blue, cyan, magenta, white) bold: Bold text dim: Dim text nl/newline: Add newline (default: True) json_key: Key for JSON output mode prefix: Optional prefix string ctx: Override context	{}

Examples:

perr("Error occurred") perr("Warning", color="yellow") perr({"error": details}, json_key="error")

Source code in provide/foundation/console/output.py

@resilient(
    fallback=None,
    suppress=(OSError, IOError, UnicodeEncodeError),
    context_provider=lambda: {"function": "perr"},
)
def perr(message: Any, **kwargs: Any) -> None:
    """Output message to stderr.

    Args:
        message: Content to output (any type - will be stringified or JSON-encoded)
        **kwargs: Optional formatting arguments:
            color: Color name (red, green, yellow, blue, cyan, magenta, white)
            bold: Bold text
            dim: Dim text
            nl/newline: Add newline (default: True)
            json_key: Key for JSON output mode
            prefix: Optional prefix string
            ctx: Override context

    Examples:
        perr("Error occurred")
        perr("Warning", color="yellow")
        perr({"error": details}, json_key="error")

    """
    ctx = kwargs.get("ctx") or _get_context()

    # Handle newline option (support both nl and newline)
    nl = kwargs.get("nl", kwargs.get("newline", True))

    if _should_use_json(ctx):
        # JSON mode
        if kwargs.get("json_key"):
            _output_json({kwargs["json_key"]: message}, sys.stderr)
        else:
            _output_json(message, sys.stderr)
    else:
        # Regular output mode
        # Add optional prefix
        output = str(message)
        if prefix := kwargs.get("prefix"):
            output = f"{prefix} {output}"

        # Apply color/formatting if requested and supported
        color = kwargs.get("color")
        bold = kwargs.get("bold", False)
        dim = kwargs.get("dim", False)

        if _HAS_CLICK:
            if (color or bold or dim) and _should_use_color(ctx, sys.stderr):
                click.secho(output, fg=color, bold=bold, dim=dim, err=True, nl=nl)
            else:
                click.echo(output, err=True, nl=nl)
        # Fallback to standard Python print
        elif nl:
            print(output, file=sys.stderr)
        else:
            print(output, file=sys.stderr, end="")

pin

pin(prompt: str = '', **kwargs: Any) -> str | Any

Input from stdin with optional prompt.

Parameters:

Name	Type	Description	Default
prompt	str	Prompt to display before input	''
**kwargs	Any	Optional formatting arguments: type: Type to convert input to (int, float, bool, etc.) default: Default value if no input provided password: Hide input for passwords (default: False) confirmation_prompt: Ask for confirmation (for passwords) hide_input: Hide the input (same as password) show_default: Show default value in prompt value_proc: Callable to process the value json_key: Key for JSON output mode ctx: Override context color: Color for prompt (red, green, yellow, blue, cyan, magenta, white) bold: Bold prompt text	{}

Returns:

Type	Description
str | Any	User input as string or converted type

Examples:

name = pin("Enter name: ") age = pin("Age: ", type=int, default=0) password = pin("Password: ", password=True)

In JSON mode, returns structured input data.

Source code in provide/foundation/console/input.py

def pin(prompt: str = "", **kwargs: Any) -> str | Any:
    """Input from stdin with optional prompt.

    Args:
        prompt: Prompt to display before input
        **kwargs: Optional formatting arguments:
            type: Type to convert input to (int, float, bool, etc.)
            default: Default value if no input provided
            password: Hide input for passwords (default: False)
            confirmation_prompt: Ask for confirmation (for passwords)
            hide_input: Hide the input (same as password)
            show_default: Show default value in prompt
            value_proc: Callable to process the value
            json_key: Key for JSON output mode
            ctx: Override context
            color: Color for prompt (red, green, yellow, blue, cyan, magenta, white)
            bold: Bold prompt text

    Returns:
        User input as string or converted type

    Examples:
        name = pin("Enter name: ")
        age = pin("Age: ", type=int, default=0)
        password = pin("Password: ", password=True)

    In JSON mode, returns structured input data.

    """
    ctx = kwargs.get("ctx") or _get_context()

    if _should_use_json(ctx):
        return _handle_json_input(prompt, kwargs)
    else:
        return _handle_interactive_input(prompt, kwargs, ctx)

pin_lines

pin_lines(count: int | None = None) -> list[str]

Read multiple lines from stdin.

Parameters:

Name	Type	Description	Default
count	int | None	Number of lines to read (None for all until EOF)	None

Returns:

Type	Description
list[str]	List of input lines

Examples:

lines = pin_lines(3) # Read exactly 3 lines all_lines = pin_lines() # Read until EOF

Source code in provide/foundation/console/input.py

def pin_lines(count: int | None = None) -> list[str]:
    """Read multiple lines from stdin.

    Args:
        count: Number of lines to read (None for all until EOF)

    Returns:
        List of input lines

    Examples:
        lines = pin_lines(3)  # Read exactly 3 lines
        all_lines = pin_lines()  # Read until EOF

    """
    lines = []
    for i, line in enumerate(pin_stream()):
        lines.append(line)
        if count is not None and i + 1 >= count:
            break
    return lines

pin_stream

pin_stream() -> Iterator[str]

Stream input line by line from stdin.

Yields:

Type	Description
str	Lines from stdin (without trailing newline)

Examples:

for line in pin_stream(): process(line)

Note: This blocks on each line. For non-blocking, use apin_stream().

Source code in provide/foundation/console/input.py

def pin_stream() -> Iterator[str]:
    """Stream input line by line from stdin.

    Yields:
        Lines from stdin (without trailing newline)

    Examples:
        for line in pin_stream():
            process(line)

    Note: This blocks on each line. For non-blocking, use apin_stream().

    """
    ctx = _get_context()

    if _should_use_json(ctx):
        # In JSON mode, try to read as JSON first
        stdin_content = sys.stdin.read()
        try:
            # Try to parse as JSON array/object
            data = json_loads(stdin_content)
            if isinstance(data, list):
                for item in data:
                    yield json_dumps(item) if not isinstance(item, str) else item
            else:
                yield json_dumps(data)
        except ValidationError:
            # Fall back to line-by-line reading
            for line in stdin_content.splitlines():
                if line:  # Skip empty lines
                    yield line
    else:
        # Regular mode - yield lines as they come
        log.debug("📥 Starting input stream")
        line_count = 0
        try:
            for line in sys.stdin:
                line = line.rstrip("\n\r")
                line_count += 1
                log.trace("📥 Stream line", line_num=line_count, length=len(line))
                yield line
        finally:
            log.debug("📥 Input stream ended", lines=line_count)

pout

pout(message: Any, **kwargs: Any) -> None

Output message to stdout.

Parameters:

Name	Type	Description	Default
message	Any	Content to output (any type - will be stringified or JSON-encoded)	required
**kwargs	Any	Optional formatting arguments: color: Color name (red, green, yellow, blue, cyan, magenta, white) bold: Bold text dim: Dim text nl/newline: Add newline (default: True) json_key: Key for JSON output mode prefix: Optional prefix string ctx: Override context	{}

Examples:

pout("Hello world") pout({"data": "value"}) # Auto-JSON if dict/list pout("Success", color="green", bold=True) pout(results, json_key="results")

Source code in provide/foundation/console/output.py

@resilient(
    fallback=None,
    suppress=(OSError, IOError, UnicodeEncodeError),
    context_provider=lambda: {"function": "pout"},
)
def pout(message: Any, **kwargs: Any) -> None:
    """Output message to stdout.

    Args:
        message: Content to output (any type - will be stringified or JSON-encoded)
        **kwargs: Optional formatting arguments:
            color: Color name (red, green, yellow, blue, cyan, magenta, white)
            bold: Bold text
            dim: Dim text
            nl/newline: Add newline (default: True)
            json_key: Key for JSON output mode
            prefix: Optional prefix string
            ctx: Override context

    Examples:
        pout("Hello world")
        pout({"data": "value"})  # Auto-JSON if dict/list
        pout("Success", color="green", bold=True)
        pout(results, json_key="results")

    """
    ctx = kwargs.get("ctx") or _get_context()

    # Handle newline option (support both nl and newline)
    nl = kwargs.get("nl", kwargs.get("newline", True))

    if _should_use_json(ctx):
        # JSON mode
        if kwargs.get("json_key"):
            _output_json({kwargs["json_key"]: message}, sys.stdout)
        else:
            _output_json(message, sys.stdout)
    else:
        # Regular output mode
        # Add optional prefix
        output = str(message)
        if prefix := kwargs.get("prefix"):
            output = f"{prefix} {output}"

        # Apply color/formatting if requested and supported
        color = kwargs.get("color")
        bold = kwargs.get("bold", False)
        dim = kwargs.get("dim", False)

        if _HAS_CLICK:
            if (color or bold or dim) and _should_use_color(ctx, sys.stdout):
                click.secho(output, fg=color, bold=bold, dim=dim, nl=nl)
            else:
                click.echo(output, nl=nl)
        # Fallback to standard Python print
        elif nl:
            print(output, file=sys.stdout)
        else:
            print(output, file=sys.stdout, end="")