Input

provide.foundation.console.input

TODO: Add module docstring.

Classes

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)

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)