Skip to content

Display

display

Display/UX helper functions for Miura API results.

This module provides convenient print functions for displaying API results in a user-friendly way. All functions print directly to stdout (or specified file).

Important: This is a library module with display helpers, NOT a CLI tool. For CLI tools, see: - miura.auth.cli - Authentication CLI tool (miura-auth command)

Layer Separation: - miura.display - Display helpers (library functions for formatting/printing) - miura.auth.cli - CLI tool (Click-based command-line application) - miura.logging - Logging infrastructure (for debugging/monitoring)

Note: This is for DISPLAYING content, not logging. For logging, use miura.logging or Python's standard logging module.

Example

from miura.display import print_items from miura.logging import get_logger

items = collection.list_items()

Display to user (print to stdout)

print_items(items)

For logging (separate concern - use miura.logging):

logger = get_logger(name) logger.info(f"Retrieved {len(items)} items") for item in items[:10]: logger.info(f" - {item}") # Uses item.str

Progress

Generic progress display: in-place status line when TTY (with optional spinner), or full lines on stdout when quiet or not a TTY.

Use for any long-running task (recipe wait, upload, download, etc.): - Call update(message) each tick; use print_line(message) for permanent lines (e.g. state transitions). Call finish() when done. - print_line() always writes to file (stdout by default), not the logger. - quiet=True: no in-place updates, no spinner; print_line() still writes to file.

Source code in src/miura/display/progress.py
class Progress:
    """
    Generic progress display: in-place status line when TTY (with optional spinner),
    or full lines on stdout when quiet or not a TTY.

    Use for any long-running task (recipe wait, upload, download, etc.):
    - Call update(message) each tick; use print_line(message) for permanent lines
      (e.g. state transitions). Call finish() when done.
    - print_line() always writes to ``file`` (stdout by default), not the logger.
    - quiet=True: no in-place updates, no spinner; print_line() still writes to file.
    """

    def __init__(
        self,
        *,
        quiet: bool = False,
        file: Optional[TextIO] = None,
        logger: Optional[Any] = None,
    ) -> None:
        self._quiet = quiet
        self._file = file if file is not None else sys.stdout
        self._logger = logger
        self._isatty = getattr(self._file, "isatty", lambda: False)()
        self._active = False  # we have an in-place line visible
        self._step = 0

    def update(self, message: str, *, spinner: bool = True) -> None:
        """
        Update the current status line in-place (TTY, not quiet).
        When quiet or not a TTY, this is a no-op; use print_line() for output.
        """
        if self._quiet or not self._isatty:
            return
        self._step += 1
        if spinner:
            line = _spinner_char(self._step) + " " + message
        else:
            line = message
        padding = max(0, 80 - len(line))
        self._file.write("\r" + line + " " * padding)
        self._file.flush()
        self._active = True

    def print_line(self, message: str) -> None:
        """
        Emit a full line (history/transition) to ``file`` (stdout by default).

        When a live in-place line is active, clears it first, then prints the message.
        """
        if self._active:
            self._file.write("\n")
            self._active = False
        self._file.write(message + "\n")
        self._file.flush()

    def finish(self) -> None:
        """End the live line (newline only). Does not print any transition line; callers print that separately."""
        if self._active:
            self._file.write("\n")
            self._file.flush()
            self._active = False

update

update(message: str, *, spinner: bool = True) -> None

Update the current status line in-place (TTY, not quiet). When quiet or not a TTY, this is a no-op; use print_line() for output.

Source code in src/miura/display/progress.py
def update(self, message: str, *, spinner: bool = True) -> None:
    """
    Update the current status line in-place (TTY, not quiet).
    When quiet or not a TTY, this is a no-op; use print_line() for output.
    """
    if self._quiet or not self._isatty:
        return
    self._step += 1
    if spinner:
        line = _spinner_char(self._step) + " " + message
    else:
        line = message
    padding = max(0, 80 - len(line))
    self._file.write("\r" + line + " " * padding)
    self._file.flush()
    self._active = True

print_line

print_line(message: str) -> None

Emit a full line (history/transition) to file (stdout by default).

When a live in-place line is active, clears it first, then prints the message.

Source code in src/miura/display/progress.py
def print_line(self, message: str) -> None:
    """
    Emit a full line (history/transition) to ``file`` (stdout by default).

    When a live in-place line is active, clears it first, then prints the message.
    """
    if self._active:
        self._file.write("\n")
        self._active = False
    self._file.write(message + "\n")
    self._file.flush()

finish

finish() -> None

End the live line (newline only). Does not print any transition line; callers print that separately.

Source code in src/miura/display/progress.py
def finish(self) -> None:
    """End the live line (newline only). Does not print any transition line; callers print that separately."""
    if self._active:
        self._file.write("\n")
        self._file.flush()
        self._active = False

LogProgressView

Default recipe wait view — module logger channel.

Source code in src/miura/display/progress_views.py
class LogProgressView:
    """Default recipe wait view — module logger channel."""

    def __init__(self, *, logger: logging.Logger | None = None, run_id_short: str | None = None) -> None:
        self._logger = logger or logging.getLogger("miura.nexus.recipe.recipe_execution")
        self._run_id_short = run_id_short

    def on_snapshot(self, snapshot: ProgressSnapshot) -> None:
        message = format_log_snapshot_line(
            snapshot.primary_text,
            snapshot.elapsed_s,
            run_id_short=None,
            is_terminal=snapshot.category == "terminal",
        )
        self._logger.log(_log_level(snapshot.severity), message)

    def on_pending_heartbeat(self, snapshot: ProgressSnapshot) -> None:
        self.on_snapshot(snapshot)

    def on_debug(self, detail: str) -> None:
        self._logger.debug("progress debug: %s", detail)

NullProgressView

No-op view for tests and headless CI.

Source code in src/miura/display/progress_views.py
class NullProgressView:
    """No-op view for tests and headless CI."""

    def on_snapshot(self, snapshot: ProgressSnapshot) -> None:
        return None

    def on_pending_heartbeat(self, snapshot: ProgressSnapshot) -> None:
        return None

    def on_debug(self, detail: str) -> None:
        return None

ProgressView

Bases: Protocol

Source code in src/miura/display/progress_views.py
@runtime_checkable
class ProgressView(Protocol):
    def on_snapshot(self, snapshot: ProgressSnapshot) -> None:
        """Render a user-facing progress snapshot."""

    def on_pending_heartbeat(self, snapshot: ProgressSnapshot) -> None:
        """Render a coarse PENDING heartbeat."""

    def on_debug(self, detail: str) -> None:
        """Render raw operator/debug detail."""

on_snapshot

on_snapshot(snapshot: ProgressSnapshot) -> None

Render a user-facing progress snapshot.

Source code in src/miura/display/progress_views.py
def on_snapshot(self, snapshot: ProgressSnapshot) -> None:
    """Render a user-facing progress snapshot."""

on_pending_heartbeat

on_pending_heartbeat(snapshot: ProgressSnapshot) -> None

Render a coarse PENDING heartbeat.

Source code in src/miura/display/progress_views.py
def on_pending_heartbeat(self, snapshot: ProgressSnapshot) -> None:
    """Render a coarse PENDING heartbeat."""

on_debug

on_debug(detail: str) -> None

Render raw operator/debug detail.

Source code in src/miura/display/progress_views.py
def on_debug(self, detail: str) -> None:
    """Render raw operator/debug detail."""

print_projects

print_projects(projects: Sequence[Union[ProjectInfo, Project]], max_items: Optional[int] = None, file: Optional[TextIO] = None) -> None

Print a list of projects.

Accepts either ProjectInfo (DTO) or Project (wrapper) objects. Both have .name and .id attributes, so duck typing works.

Source code in src/miura/display/printers.py
def print_projects(
    projects: Sequence[Union[ProjectInfo, Project]], max_items: Optional[int] = None, file: Optional[TextIO] = None
) -> None:
    """
    Print a list of projects.

    Accepts either ProjectInfo (DTO) or Project (wrapper) objects.
    Both have .name and .id attributes, so duck typing works.
    """
    console = create_console(file=file)

    if console:
        primary_color = get_rich_style("primary")
        muted_color = get_rich_style("muted")
        count = len(projects)
        console.print(f"[bold]Found {count} project{'s' if count != 1 else ''}[/bold]")
        if projects:
            console.print("[bold]Projects:[/bold]")
            display_items = projects[:max_items] if max_items is not None else projects
            for project in display_items:
                console.print(
                    f"  [{primary_color}]{project.name}[/{primary_color}] "
                    f"[{muted_color}]({project.id})[/{muted_color}]"
                )
            if max_items is not None and len(projects) > max_items:
                console.print(
                    f"  [{muted_color}]... (showing first {max_items} of {len(projects)} projects)[/{muted_color}]"
                )
    else:
        # Fallback to plain text
        if file is None:
            file = sys.stdout
        print(_format_projects(projects, max_items=max_items), file=file)

print_collections

print_collections(collections: Sequence[Union[CollectionInfo, Collection]], max_items: Optional[int] = None, file: Optional[TextIO] = None) -> None

Print a list of collections.

Accepts either CollectionInfo (DTO) or Collection (wrapper) objects. Both have .name and .id attributes, so duck typing works.

Source code in src/miura/display/printers.py
def print_collections(
    collections: Sequence[Union[CollectionInfo, Collection]],
    max_items: Optional[int] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print a list of collections.

    Accepts either CollectionInfo (DTO) or Collection (wrapper) objects.
    Both have .name and .id attributes, so duck typing works.
    """
    console = create_console(file=file)

    if console:
        primary_color = get_rich_style("primary")
        muted_color = get_rich_style("muted")
        count = len(collections)
        console.print(f"[bold]Found {count} collection{'s' if count != 1 else ''}[/bold]")
        if collections:
            console.print("[bold]Collections:[/bold]")
            display_items = collections[:max_items] if max_items is not None else collections
            for collection in display_items:
                console.print(
                    f"  [{primary_color}]{collection.name}[/{primary_color}] "
                    f"[{muted_color}]({collection.id})[/{muted_color}]"
                )
            if max_items is not None and len(collections) > max_items:
                console.print(
                    f"  [{muted_color}]... (showing first {max_items} of "
                    f"{len(collections)} collections)[/{muted_color}]"
                )
    else:
        # Fallback to plain text
        if file is None:
            file = sys.stdout
        print(_format_collections(collections, max_items=max_items), file=file)

print_items

print_items(items: Sequence[CollectionItemInfo] | Iterable[CollectionItemInfo], style: Literal['list', 'table'] = 'list', max_items: Optional[int] = None, show_attributes: bool = False, file: Optional[TextIO] = None) -> None

Print flat items for display.

Use this for items from list_items() or iter_items().

For items with depth information from walk_items(), use print_walk_items() instead.

Examples:

From list_items()

items = collection.list_items() print_items(items)

From iter_items()

print_items(collection.iter_items(), max_items=50)

Include split/labels/annotations from item attributes

print_items(items, show_attributes=True)

Source code in src/miura/display/printers.py
def print_items(
    items: Sequence[CollectionItemInfo] | Iterable[CollectionItemInfo],
    style: Literal["list", "table"] = "list",
    max_items: Optional[int] = None,
    show_attributes: bool = False,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print flat items for display.

    Use this for items from `list_items()` or `iter_items()`.

    For items with depth information from `walk_items()`, use `print_walk_items()` instead.

    Examples:
        # From list_items()
        items = collection.list_items()
        print_items(items)

        # From iter_items()
        print_items(collection.iter_items(), max_items=50)

        # Include split/labels/annotations from item attributes
        print_items(items, show_attributes=True)
    """
    console = create_console(file=file)

    # Convert to list for counting and slicing
    items_list = list(items)
    total_count = len(items_list)
    display_items = items_list[:max_items] if max_items is not None else items_list

    # Count files and folders separately
    file_count = sum(1 for item in items_list if not item.is_folder)
    folder_count = sum(1 for item in items_list if item.is_folder)

    # Build count message with proper singular/plural
    if file_count == 0 and folder_count == 0:
        count_msg = "Found 0 items"
    elif file_count == 0:
        count_msg = f"Found {folder_count} folder{'s' if folder_count != 1 else ''}"
    elif folder_count == 0:
        count_msg = f"Found {file_count} file{'s' if file_count != 1 else ''}"
    else:
        count_msg = (
            f"Found {file_count} file{'s' if file_count != 1 else ''}, "
            f"{folder_count} folder{'s' if folder_count != 1 else ''}"
        )

    if console:
        console.print(f"[bold]{count_msg}[/bold]")
        if display_items:
            if style == "table":
                # Use rich Table for table style
                primary_color = get_rich_style("primary")
                warning_color = get_rich_style("warning")
                success_color = get_rich_style("success")
                muted_color = get_rich_style("muted")
                show_type_column = any(item.is_folder for item in display_items)
                table = Table(show_header=True, header_style="bold", box=box.ROUNDED)
                table.add_column("Name", style=primary_color)
                if show_type_column:
                    table.add_column("Type", style=warning_color)
                table.add_column("Size", style=success_color, justify="right")
                if show_attributes:
                    table.add_column("Split", style=warning_color)
                    table.add_column("Labels", style=muted_color)
                    table.add_column("Annotations", style=muted_color)
                for item in display_items:
                    item_type = "Folder" if item.is_folder else "File"
                    size_str = human_bytes(item.file_size) if item.file_size else "-"
                    uri = item.item_uri.rstrip("/") + "/" if item.is_folder else item.item_uri
                    row = [uri]
                    if show_type_column:
                        row.append(item_type)
                    row.append(size_str)
                    if show_attributes:
                        split, labels, annotations = _item_attribute_summary_rich(_item_raw_attributes(item))
                        row.extend([split, labels, annotations])
                    table.add_row(*row)
                console.print(table)
            else:
                # Use rich list style
                folder_color = get_rich_style("folder")
                file_color = get_rich_style("file")
                muted_color = get_rich_style("muted")
                for item in display_items:
                    if item.is_folder:
                        uri = item.item_uri.rstrip("/") + "/"
                        console.print(f"  [{folder_color}]{uri}[/{folder_color}]")
                    else:
                        size_str = (
                            f" [{muted_color}]({human_bytes(item.file_size)})[/{muted_color}]" if item.file_size else ""
                        )
                        console.print(f"  [{file_color}]{item.item_uri}[/{file_color}]{size_str}")
                    if show_attributes:
                        split, labels, annotations = _item_attribute_summary_rich(_item_raw_attributes(item))
                        if split != "-":
                            line = Text("    ")
                            line.append("split", style=muted_color)
                            line.append(": ")
                            line.append(split)
                            console.print(line)
                        if labels != "-":
                            line = Text("    ")
                            line.append("labels", style=muted_color)
                            line.append(": ")
                            line.append(labels)
                            console.print(line)
                        if annotations != "-":
                            line = Text("    ")
                            line.append("annotations", style=muted_color)
                            line.append(": ")
                            line.append(annotations)
                            console.print(line)
            if max_items is not None and total_count > max_items:
                muted_color = get_rich_style("muted")
                console.print(
                    f"  [{muted_color}]... (showing first {max_items} of {total_count} items)[/{muted_color}]"
                )
    else:
        # Fallback to plain text
        if file is None:
            file = sys.stdout
        print(_format_items(items, style=style, max_items=max_items), file=file)
        if show_attributes and display_items:
            for item in display_items:
                split, labels, annotations = _item_attribute_summary(_item_raw_attributes(item))
                if split != "-" or labels != "-" or annotations != "-":
                    print(f"  {item.item_uri}", file=file)
                    if split != "-":
                        print(f"    split: {split}", file=file)
                    if labels != "-":
                        print(f"    labels: {labels}", file=file)
                    if annotations != "-":
                        print(f"    annotations: {annotations}", file=file)

print_walk_items

print_walk_items(items: Union[Iterable[tuple[CollectionItemInfo, int]], AsyncIterable[tuple[Any, int]]], max_items: Optional[int] = None, file: Optional[TextIO] = None, collection_name: str | None = None) -> None

Print items from walk_items() with depth-aware formatting.

This function accepts both sync and async iterators from walk_items(). For async iterators, items are collected first, then processed.

Parameters:

Name Type Description Default
items Union[Iterable[tuple[CollectionItemInfo, int]], AsyncIterable[tuple[Any, int]]]

Iterable or AsyncIterable of (item, depth) tuples from walk_items()

required
max_items Optional[int]

Maximum number of items to display (None = no limit)

None
file Optional[TextIO]

File to write to (default: stdout)

None
collection_name str | None

Optional collection name to use as root node (default: "Collection")

None

Examples:

From walk_items() (sync) - no limits

print_walk_items(collection.walk_items())

From walk_items() (async) - accepts async iterator directly

print_walk_items(collection.walk_items(), collection_name=collection.name)

With optional limits

print_walk_items(collection.walk_items(max_depth=3), max_items=50)

Source code in src/miura/display/printers.py
def print_walk_items(
    items: Union[Iterable[tuple[CollectionItemInfo, int]], AsyncIterable[tuple[Any, int]]],
    max_items: Optional[int] = None,
    file: Optional[TextIO] = None,
    collection_name: str | None = None,
) -> None:
    """
    Print items from walk_items() with depth-aware formatting.

    This function accepts both sync and async iterators from walk_items().
    For async iterators, items are collected first, then processed.

    Args:
        items: Iterable or AsyncIterable of (item, depth) tuples from walk_items()
        max_items: Maximum number of items to display (None = no limit)
        file: File to write to (default: stdout)
        collection_name: Optional collection name to use as root node (default: "Collection")

    Examples:
        # From walk_items() (sync) - no limits
        print_walk_items(collection.walk_items())

        # From walk_items() (async) - accepts async iterator directly
        print_walk_items(collection.walk_items(), collection_name=collection.name)

        # With optional limits
        print_walk_items(collection.walk_items(max_depth=3), max_items=50)
    """
    # Handle async iterators by collecting them first
    if hasattr(items, "__aiter__"):
        # It's an async iterator - collect items first
        async def _collect_async_items():
            collected = []
            async for item, depth in items:  # type: ignore
                collected.append((item, depth))
            return collected

        # Check if we're in a running event loop
        try:
            loop = asyncio.get_running_loop()
            # We're in an async context - use nest_asyncio to allow nested event loops
            import nest_asyncio

            nest_asyncio.apply(loop)
            # Now we can use run_until_complete even though loop is running
            items_list: list[tuple[CollectionItemInfo, int]] = loop.run_until_complete(_collect_async_items())
        except RuntimeError:
            # No running event loop - we can use asyncio.run()
            items_list = asyncio.run(_collect_async_items())
    else:
        # It's a sync iterable - convert to list for processing
        items_list = list(items)  # type: ignore[arg-type]

    console = create_console(file=file)

    if console:
        # Create a single root tree node for the collection
        root_label = collection_name if collection_name else "Collection"
        root_tree = Tree(f"[bold]{root_label}[/bold]")

        # Use rich Tree for hierarchical display
        count = 0
        file_count = 0
        folder_count = 0
        parent_paths: dict[int, str] = {}
        tree_nodes: dict[int, Tree] = {}  # Track tree nodes at each depth (relative to root)

        for item, depth in items_list:
            # Count files and folders during iteration
            if item.is_folder:
                folder_count += 1
            else:
                file_count += 1

            if item.is_folder:
                # Compute relative path
                if depth == 0:
                    uri = item.item_uri.rstrip("/") + "/"
                    display_path = uri
                else:
                    parent_uri = parent_paths.get(depth - 1, "")
                    if parent_uri and item.item_uri.startswith(parent_uri):
                        relative_path = item.item_uri[len(parent_uri) :].lstrip("/")
                        display_path = relative_path.rstrip("/") + "/"
                    else:
                        display_path = item.item_uri.rstrip("/") + "/"

                # Create or get tree node (all items are children of root_tree)
                folder_color = get_rich_style("folder")
                if depth == 0:
                    # Root level item - add directly to root tree
                    new_node = root_tree.add(f"[{folder_color}]{display_path}[/{folder_color}]")
                    tree_nodes[depth] = new_node
                else:
                    # Child level - add to parent tree node
                    parent_node = tree_nodes.get(depth - 1)
                    if parent_node:
                        new_node = parent_node.add(f"[{folder_color}]{display_path}[/{folder_color}]")
                        tree_nodes[depth] = new_node
                    else:
                        # Fallback: add to root if parent not found
                        new_node = root_tree.add(f"[{folder_color}]{display_path}[/{folder_color}]")
                        tree_nodes[depth] = new_node

                parent_paths[depth] = item.item_uri.rstrip("/") + "/"
            else:
                # File item
                file_color = get_rich_style("file")
                muted_color = get_rich_style("muted")
                if depth == 0:
                    # Root level file - add directly to root tree
                    size_str = (
                        f" [{muted_color}]({human_bytes(item.file_size)})[/{muted_color}]" if item.file_size else ""
                    )
                    display_path = item.item_uri
                    root_tree.add(f"[{file_color}]{display_path}[/{file_color}]{size_str}")
                else:
                    # Child file: show relative path
                    parent_uri = parent_paths.get(depth - 1, "")
                    if parent_uri and item.item_uri.startswith(parent_uri):
                        relative_path = item.item_uri[len(parent_uri) :].lstrip("/")
                    else:
                        relative_path = item.item_uri

                    size_str = (
                        f" [{muted_color}]({human_bytes(item.file_size)})[/{muted_color}]" if item.file_size else ""
                    )
                    parent_node = tree_nodes.get(depth - 1)
                    if parent_node:
                        parent_node.add(f"[{file_color}]{relative_path}[/{file_color}]{size_str}")
                    else:
                        # Fallback: add to root if parent not found
                        root_tree.add(f"[{file_color}]{relative_path}[/{file_color}]{size_str}")

            count += 1
            if max_items is not None and count >= max_items:
                break

        # Build count message with proper singular/plural
        if file_count == 0 and folder_count == 0:
            count_msg = "Found 0 items"
        elif file_count == 0:
            count_msg = f"Found {folder_count} folder{'s' if folder_count != 1 else ''}"
        elif folder_count == 0:
            count_msg = f"Found {file_count} file{'s' if file_count != 1 else ''}"
        else:
            count_msg = (
                f"Found {file_count} file{'s' if file_count != 1 else ''}, "
                f"{folder_count} folder{'s' if folder_count != 1 else ''}"
            )

        # Print header and tree
        console.print(f"[bold]{count_msg}[/bold]")
        if count > 0:
            console.print(root_tree)
            if max_items is not None and count >= max_items:
                muted_color = get_rich_style("muted")
                console.print(f"  [{muted_color}]... (showing first {max_items} items)[/{muted_color}]")
    else:
        # Fallback to plain text
        if file is None:
            file = sys.stdout
        print(_format_walk_items(items_list, max_items=max_items), file=file)

print_collection_schema

print_collection_schema(schema: List[Dict[str, Any]], file: Optional[TextIO] = None) -> None

Print a collection schema with standard JSON formatting and raw strings for patterns.

Parameters:

Name Type Description Default
schema List[Dict[str, Any]]

List of schema node dictionaries

required
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/printers.py
def print_collection_schema(schema: List[Dict[str, Any]], file: Optional[TextIO] = None) -> None:
    """
    Print a collection schema with standard JSON formatting and raw strings for patterns.

    Args:
        schema: List of schema node dictionaries
        file: Optional file to write to (defaults to stdout)
    """
    console = create_console(file=file)

    def format_node(node: Dict[str, Any], indent: int = 0, use_rich: bool = False) -> str:
        """Format a single schema node with standard JSON formatting."""
        indent_str = " " * indent
        field_indent = " " * (indent + 4)  # Fields are indented 4 more than the object
        items = []
        for key, value in node.items():
            if key == "pattern" and isinstance(value, str):
                # Use raw string for patterns
                if use_rich:
                    pattern_color = get_rich_style("primary")
                    items.append(f'{field_indent}"{key}": [{pattern_color}]r"{value}"[/{pattern_color}]')
                else:
                    items.append(f'{field_indent}"{key}": r"{value}"')
            elif key == "name" and isinstance(value, str):
                if use_rich:
                    name_color = get_rich_style("file")
                    items.append(f'{field_indent}"{key}": [{name_color}]"{value}"[/{name_color}]')
                else:
                    items.append(f'{field_indent}"{key}": "{value}"')
            elif key == "children" and isinstance(value, list):
                children_str = format_children(value, indent + 4, use_rich)
                items.append(f'{field_indent}"{key}": {children_str}')
            else:
                if use_rich:
                    muted_color = get_rich_style("muted")
                    items.append(f'{field_indent}"{key}": [{muted_color}]{value}[/{muted_color}]')
                else:
                    items.append(f'{field_indent}"{key}": {value}')
        return indent_str + "{\n" + ",\n".join(items) + "\n" + indent_str + "}"

    def format_children(children: List[Dict[str, Any]], indent: int = 0, use_rich: bool = False) -> str:
        """Format children list with standard JSON formatting."""
        if not children:
            return "[]"
        indent_str = " " * indent
        formatted_nodes = []
        for child in children:
            # Format node with proper indentation (indent + 4 for array content)
            node_str = format_node(child, indent + 4, use_rich)
            formatted_nodes.append(node_str)
        # Standard JSON format: each object on its own line, properly indented
        return "[\n" + ",\n".join(formatted_nodes) + "\n" + indent_str + "]"

    if console:
        # Use Rich formatting
        if not schema:
            console.print("[bold]Schema:[/bold] []")
            return

        console.print("[bold]Schema:[/bold]")
        formatted_schema = format_children(schema, indent=0, use_rich=True)
        console.print(formatted_schema)
    else:
        # Fallback to plain text
        if file is None:
            file = sys.stdout
        if not schema:
            print("Schema: []", file=file)
            return
        print("Schema:", file=file)
        formatted_schema = format_children(schema, indent=0, use_rich=False)
        print(formatted_schema, file=file)

print_recipes

print_recipes(recipes: Sequence[RecipeResponse], max_items: Optional[int] = None, file: Optional[TextIO] = None) -> None

Print a list of recipes.

Parameters:

Name Type Description Default
recipes Sequence[RecipeResponse]

Sequence of RecipeResponse objects

required
max_items Optional[int]

Optional maximum number of items to display

None
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/printers.py
def print_recipes(
    recipes: Sequence["RecipeResponse"],
    max_items: Optional[int] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print a list of recipes.

    Args:
        recipes: Sequence of RecipeResponse objects
        max_items: Optional maximum number of items to display
        file: Optional file to write to (defaults to stdout)
    """
    # Lazy import to avoid circular dependency
    from miura.nexus.recipe.recipe_models import RecipeResponse  # noqa: F401

    console = create_console(file=file)

    if console:
        primary_color = get_rich_style("primary")
        muted_color = get_rich_style("muted")
        count = len(recipes)
        console.print(f"[bold]Found {count} recipe{'s' if count != 1 else ''}[/bold]")
        if recipes:
            console.print("[bold]Recipes:[/bold]")
            display_items = recipes[:max_items] if max_items is not None else recipes
            for recipe in display_items:
                # Try to get source/target info, handle ValidationError gracefully
                try:
                    source = f"{recipe.source_collection}:{recipe.source_group}"
                    try:
                        target = f"{recipe.target_collection}:{recipe.target_group}"
                    except Exception:
                        target = "[dim]<omitted>[/dim]"
                    console.print(
                        f"  [{primary_color}]{recipe.name}[/{primary_color}] "
                        f"[{muted_color}]({recipe.id})[/{muted_color}]"
                    )
                    console.print(f"    Source: {source}, Target: {target}")
                except Exception:
                    # Fallback if convenience properties fail
                    console.print(
                        f"  [{primary_color}]{recipe.name}[/{primary_color}] "
                        f"[{muted_color}]({recipe.id})[/{muted_color}]"
                    )
            if max_items is not None and len(recipes) > max_items:
                console.print(
                    f"  [{muted_color}]... (showing first {max_items} of " f"{len(recipes)} recipes)[/{muted_color}]"
                )
    else:
        # Fallback to plain text
        if file is None:
            file = sys.stdout
        print(_format_recipes(recipes, max_items=max_items), file=file)

print_project_info

print_project_info(project: ProjectInfo, collections: Optional[list[CollectionInfo]] = None, recipes: Optional[list[RecipeResponse]] = None, file: Optional[TextIO] = None) -> None

Print comprehensive information about a project, its collections, and recipes.

Parameters:

Name Type Description Default
project ProjectInfo

ProjectInfo object with project details

required
collections Optional[list[CollectionInfo]]

Optional list of CollectionInfo objects (if None, will show message)

None
recipes Optional[list[RecipeResponse]]

Optional list of RecipeResponse objects (if None, will show message)

None
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_project_info(
    project: ProjectInfo,
    collections: Optional[list[CollectionInfo]] = None,
    recipes: Optional[list["RecipeResponse"]] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print comprehensive information about a project, its collections, and recipes.

    Args:
        project: ProjectInfo object with project details
        collections: Optional list of CollectionInfo objects (if None, will show message)
        recipes: Optional list of RecipeResponse objects (if None, will show message)
        file: Optional file to write to (defaults to stdout)
    """
    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")
    error_color = get_rich_style("error")

    if console:
        console.print("[bold]PROJECT INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print()
        console.print("[bold]Project Details:[/bold]")
        console.print(f"   Name: [{primary_color}]{project.name}[/{primary_color}]")
        console.print(f"   UUID: [{muted_color}]{project.id}[/{muted_color}]")
        project_metadata = _normalized_metadata(project.metadata)
        owner_line = format_project_owner_line(
            project.owner_id,
            owner_profile=project.owner_profile,
            metadata=project_metadata,
        )
        if owner_line:
            console.print(f"   Owner: [{muted_color}]{owner_line}[/{muted_color}]")
        project_created, project_updated = _display_timestamps(project.created_at, project.updated_at, project_metadata)
        if project_created:
            console.print(f"   Created: [{muted_color}]{project_created}[/{muted_color}]")
        if project_updated:
            console.print(f"   Updated: [{muted_color}]{project_updated}[/{muted_color}]")

        # Handle metadata
        project_metadata_display = _metadata_for_display(project_metadata)
        if project_metadata_display:
            console.print("   Metadata:")
            _print_metadata_value(
                project_metadata_display,
                indent="     ",
                console=console,
                primary_color=primary_color,
                muted_color=muted_color,
            )

        # Collections information
        console.print()
        console.print("[bold]COLLECTIONS INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print()

        if collections is None:
            console.print(f"[{error_color}]Error: Could not list collections[/{error_color}]")
        elif not collections:
            console.print(f"[{muted_color}]No collections found in this project[/{muted_color}]")
        else:
            console.print(f"Found {len(collections)} collection{'s' if len(collections) != 1 else ''}:")
            console.print("[dim]" + "-" * 60 + "[/dim]")
            console.print()

            for i, collection in enumerate(collections, 1):
                console.print(f"{i}. [{primary_color}]{collection.name}[/{primary_color}]")
                console.print(f"   UUID: [{muted_color}]{collection.uuid}[/{muted_color}]")

                # Handle collection attributes + remaining metadata
                collection_metadata = _normalized_metadata(collection.metadata)
                collection_created, collection_updated = _display_timestamps(
                    collection.created_at, collection.updated_at, collection_metadata
                )
                collection_attributes_display, collection_metadata_display = _collection_fields_for_display(
                    collection_metadata
                )
                if collection_attributes_display:
                    console.print("   Attributes:")
                    _print_metadata_value(
                        collection_attributes_display,
                        indent="     ",
                        console=console,
                        primary_color=primary_color,
                        muted_color=muted_color,
                    )
                if collection_metadata_display:
                    console.print("   Metadata:")
                    _print_metadata_value(
                        collection_metadata_display,
                        indent="     ",
                        console=console,
                        primary_color=primary_color,
                        muted_color=muted_color,
                    )

                # Handle timestamps
                if collection_created:
                    console.print(f"   Created: [{muted_color}]{collection_created}[/{muted_color}]")
                if collection_updated:
                    console.print(f"   Updated: [{muted_color}]{collection_updated}[/{muted_color}]")

                console.print("[dim]" + "-" * 40 + "[/dim]")
                console.print()

        # Recipes information
        console.print()
        console.print("[bold]RECIPES INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print()
        if recipes is None:
            console.print(f"[{error_color}]Error: Could not list recipes[/{error_color}]")
        elif not recipes:
            console.print(f"[{muted_color}]No recipes found in this project[/{muted_color}]")
        else:
            console.print(f"Found {len(recipes)} recipe{'s' if len(recipes) != 1 else ''}:")
            console.print("[dim]" + "-" * 60 + "[/dim]")
            console.print()
            for i, recipe in enumerate(recipes, 1):
                console.print(f"{i}. [{primary_color}]{recipe.name}[/{primary_color}]")
                console.print(f"   UUID: [{muted_color}]{recipe.id}[/{muted_color}]")
                console.print(f"   Created by: [{muted_color}]{recipe.created_by}[/{muted_color}]")
                console.print(f"   Created at: [{muted_color}]{recipe.created_at}[/{muted_color}]")
                if recipe.description:
                    console.print(f"   description: [{muted_color}]{recipe.description}[/{muted_color}]")

                # Sources information
                console.print("   [bold]Sources:[/bold]")
                if recipe.sources:
                    for i, source in enumerate(recipe.sources, 1):
                        console.print(
                            f"      {i}. [{primary_color}]{source.collection_name}[/{primary_color}] "
                            f"(group: [{muted_color}]{source.group_name}[/{muted_color}])"
                        )
                else:
                    console.print(f"      [{muted_color}]No sources (data integrity issue)[/{muted_color}]")

                # Targets information
                console.print("   [bold]Targets:[/bold]")
                if recipe.targets:
                    for i, target in enumerate(recipe.targets, 1):
                        console.print(
                            f"      {i}. [{primary_color}]{target.collection_name}[/{primary_color}] "
                            f"(group: [{muted_color}]{target.group_name}[/{muted_color}])"
                        )
                else:
                    console.print(f"      [{muted_color}]Omitted (will be created during execution)[/{muted_color}]")

                console.print("[dim]" + "-" * 40 + "[/dim]")
                console.print()
    else:
        # Plain text fallback
        if file is None:
            file = sys.stdout
        print("PROJECT INFORMATION", file=file)
        print("=" * 60, file=file)
        print("Project Details:", file=file)
        print(f"   Name: {project.name}", file=file)
        print(f"   UUID: {project.id}", file=file)
        project_metadata = _normalized_metadata(project.metadata)
        owner_line = format_project_owner_line(
            project.owner_id,
            owner_profile=project.owner_profile,
            metadata=project_metadata,
        )
        if owner_line:
            print(f"   Owner: {owner_line}", file=file)
        project_created, project_updated = _display_timestamps(project.created_at, project.updated_at, project_metadata)
        if project_created:
            print(f"   Created: {project_created}", file=file)
        if project_updated:
            print(f"   Updated: {project_updated}", file=file)
        project_metadata_display = _metadata_for_display(project_metadata)
        if project_metadata_display:
            print("   Metadata:", file=file)
            _print_metadata_value(project_metadata_display, indent="     ", file=file)
        print("\nCOLLECTIONS INFORMATION", file=file)
        print("=" * 60, file=file)
        if collections is None:
            print("Error: Could not list collections", file=file)
        elif not collections:
            print("No collections found in this project", file=file)
        else:
            print(f"Found {len(collections)} collection(s):", file=file)
            print("-" * 60, file=file)
            for i, collection in enumerate(collections, 1):
                print(f"{i}. {collection.name}", file=file)
                print(f"   UUID: {collection.uuid}", file=file)
                collection_metadata = _normalized_metadata(collection.metadata)
                collection_created, collection_updated = _display_timestamps(
                    collection.created_at, collection.updated_at, collection_metadata
                )
                collection_attributes_display, collection_metadata_display = _collection_fields_for_display(
                    collection_metadata
                )
                if collection_attributes_display:
                    print("   Attributes:", file=file)
                    _print_metadata_value(collection_attributes_display, indent="     ", file=file)
                if collection_metadata_display:
                    print("   Metadata:", file=file)
                    _print_metadata_value(collection_metadata_display, indent="     ", file=file)
                if collection_created:
                    print(f"   Created: {collection_created}", file=file)
                if collection_updated:
                    print(f"   Updated: {collection_updated}", file=file)
                print("-" * 40, file=file)

        print("\nRECIPES INFORMATION", file=file)
        print("=" * 60, file=file)
        if recipes is None:
            print("Error: Could not list recipes", file=file)
        elif not recipes:
            print("No recipes found in this project", file=file)
        else:
            print(f"Found {len(recipes)} recipe(s):", file=file)
            print("-" * 60, file=file)
            for i, recipe in enumerate(recipes, 1):
                print(f"{i}. {recipe.name}", file=file)
                print(f"   UUID: {recipe.id}", file=file)
                print("-" * 40, file=file)

print_project_help

print_project_help(file: Optional[TextIO] = None) -> None

Print help information for Project API methods.

Parameters:

Name Type Description Default
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_project_help(file: Optional[TextIO] = None) -> None:
    """
    Print help information for Project API methods.

    Args:
        file: Optional file to write to (defaults to stdout)
    """
    console = create_console(file=file)
    muted_color = get_rich_style("muted")
    accent_color = get_rich_style("accent")

    help_text = [
        ("Project API - Available Methods", "bold"),
        ("=" * 60, "dim"),
        ("", None),
        ("Collection Management:", "bold"),
        ("  list_collections() -> list[CollectionInfo]", None),
        ("    Example: collections = project.list_collections()", None),
        ("    Note: Returns CollectionInfo dataclass objects (use collection.name, collection.id)", None),
        ("", None),
        ("  get_collection(collection_name: str) -> Collection", None),
        ("    Example: collection = project.get_collection('my-collection')", None),
        ("    Note: Returns Collection wrapper with .name, .id attributes", None),
        ("", None),
        ("  create_collection(name: str, schema: dict) -> Collection", None),
        ("    Example: collection = project.create_collection('my-collection', {...})", None),
        ("    Note: Creates new collection and returns Collection wrapper", None),
        ("", None),
        ("  iter_collections(prefetch_pages=1, page_size=50) -> Iterator[CollectionInfo]", None),
        ("    Example: for collection in project.iter_collections(): ...", None),
        ("    Note: Yields CollectionInfo dataclass objects (use collection.name, collection.id)", None),
        ("", None),
        ("Project Operations:", "bold"),
        ("  update_project(metadata=None) -> Project", None),
        ("    Example: project.update_project({'key': 'value'})  # Not yet implemented", None),
        ("", None),
        ("  delete_project() -> None", None),
        ("    Example: project.delete_project()", None),
        ("    Note: Deletes the project and all associated resources (collections, data, etc.)", None),
        ("", None),
        ("Information:", "bold"),
        ("  print_info() -> None", None),
        ("    Example: project.print_info()", None),
    ]

    if console:
        for text, style in help_text:
            if style == "bold":
                console.print(f"[bold]{text}[/bold]")
            elif style == "dim":
                console.print(f"[dim]{text}[/dim]")
            elif text.startswith("  ") and "()" in text:
                # Method signature - use accent color
                method_part = text.split(" -> ")[0]
                return_part = text.split(" -> ")[1] if " -> " in text else ""
                if return_part:
                    console.print(f"  [{accent_color}]{method_part}[/{accent_color}] -> {return_part}")
                else:
                    console.print(f"  [{accent_color}]{method_part}[/{accent_color}]")
            elif text.startswith("    Example:") or text.startswith("    Note:"):
                # Examples and notes - use muted color
                console.print(f"[{muted_color}]{text}[/{muted_color}]")
            else:
                console.print(text)
    else:
        # Plain text fallback
        if file is None:
            file = sys.stdout
        for text, _ in help_text:
            print(text, file=file)

print_collection_info

print_collection_info(collection: CollectionInfo, file: Optional[TextIO] = None) -> None

Print comprehensive information about a collection.

Parameters:

Name Type Description Default
collection CollectionInfo

CollectionInfo object with collection details

required
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_collection_info(
    collection: CollectionInfo,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print comprehensive information about a collection.

    Args:
        collection: CollectionInfo object with collection details
        file: Optional file to write to (defaults to stdout)
    """
    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")

    if console:
        console.print("[bold]COLLECTION INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print()
        console.print("[bold]Collection Details:[/bold]")
        console.print(f"   Name: [{primary_color}]{collection.name}[/{primary_color}]")
        console.print(f"   UUID: [{muted_color}]{collection.uuid}[/{muted_color}]")
        console.print(f"   Project UUID: [{muted_color}]{collection.project_id}[/{muted_color}]")
        collection_metadata = _normalized_metadata(collection.metadata)
        collection_created, collection_updated = _display_timestamps(
            collection.created_at, collection.updated_at, collection_metadata
        )
        if collection_created:
            console.print(f"   Created: [{muted_color}]{collection_created}[/{muted_color}]")
        if collection_updated:
            console.print(f"   Updated: [{muted_color}]{collection_updated}[/{muted_color}]")

        # Handle metadata
        collection_attributes_display, collection_metadata_display = _collection_fields_for_display(collection_metadata)
        if collection_attributes_display:
            console.print("   Attributes:")
            _print_metadata_value(
                collection_attributes_display,
                indent="     ",
                console=console,
                primary_color=primary_color,
                muted_color=muted_color,
            )
        if collection_metadata_display:
            console.print("   Metadata:")
            _print_metadata_value(
                collection_metadata_display,
                indent="     ",
                console=console,
                primary_color=primary_color,
                muted_color=muted_color,
            )

        # Items information
        console.print()
        console.print("[bold]ITEMS INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
    else:
        # Plain text fallback
        if file is None:
            file = sys.stdout
        print("COLLECTION INFORMATION", file=file)
        print("=" * 60, file=file)
        print("Collection Details:", file=file)
        print(f"   Name: {collection.name}", file=file)
        print(f"   UUID: {collection.uuid}", file=file)
        print(f"   Project UUID: {collection.project_id}", file=file)
        collection_metadata = _normalized_metadata(collection.metadata)
        collection_created, collection_updated = _display_timestamps(
            collection.created_at, collection.updated_at, collection_metadata
        )
        if collection_created:
            print(f"   Created: {collection_created}", file=file)
        if collection_updated:
            print(f"   Updated: {collection_updated}", file=file)
        collection_attributes_display, collection_metadata_display = _collection_fields_for_display(collection_metadata)
        if collection_attributes_display:
            print("   Attributes:", file=file)
            _print_metadata_value(collection_attributes_display, indent="     ", file=file)
        if collection_metadata_display:
            print("   Metadata:", file=file)
            _print_metadata_value(collection_metadata_display, indent="     ", file=file)
        print("\nITEMS INFORMATION", file=file)
        print("=" * 60, file=file)

print_collection_help

print_collection_help(file: Optional[TextIO] = None) -> None

Print help information for Collection API methods.

Parameters:

Name Type Description Default
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_collection_help(file: Optional[TextIO] = None) -> None:
    """
    Print help information for Collection API methods.

    Args:
        file: Optional file to write to (defaults to stdout)
    """
    console = create_console(file=file)
    muted_color = get_rich_style("muted")
    accent_color = get_rich_style("accent")

    help_text = [
        ("Collection API - Available Methods", "bold"),
        ("=" * 60, "dim"),
        ("", None),
        ("Item Management:", "bold"),
        ("  list_items(path='/', sort_by=None, sort_order='asc') -> List[CollectionItemInfo]", None),
        ("    Example: items = collection.list_items(path='/data')", None),
        ("", None),
        (
            (
                "  iter_items(path='/', sort_by=None, sort_order='asc', "
                "prefetch_pages=1, page_size=50) -> Iterator[CollectionItemInfo]"
            ),
            None,
        ),
        ("    Example: for item in collection.iter_items(path='/data'): ...", None),
        ("", None),
        ("  walk_items(path='/', max_depth=None, topdown=True) -> Iterator[tuple[CollectionItemInfo, int]]", None),
        ("    Example: for item, depth in collection.walk_items(path='/data'): ...", None),
        ("", None),
        ("Data Operations:", "bold"),
        ("  plan(datasource, mode=..., case_grouping=...) -> IngestionPlan  then  plan.apply() -> UploadResult", None),
        ("    Example: plan = await collection.plan(datasource); result = await plan.apply()", None),
        ("  upload(datasource, mode=..., case_grouping=...) -> UploadResult  (plan-first wrapper)", None),
        ("    Example: result = await collection.upload(datasource, case_grouping=grouping)", None),
        ("", None),
        ("  fetch_validation_report on IngestionPlan for gate failure detail (IS manifest)", None),
        ("", None),
        ("  iter_items / walk_items / list_items -> typed browse (canonical read path)", None),
        ("    Example: async for item in collection.iter_items(): ...", None),
        ("", None),
        ("  download(path: str, local_path: str, confirm=True) -> DownloadResult", None),
        ("    Example: result = await collection.download('/data', './downloads/')", None),
        ("", None),
        ("Collection Operations:", "bold"),
        ("  set_attributes(**attrs) -> Collection", None),
        ("    Example: collection.set_attributes(dataset_stage='curated', owner_team='ml-platform')", None),
        ("", None),
        ("  get_attributes() -> dict[str, Any]", None),
        ("    Example: attrs = collection.get_attributes()", None),
        ("", None),
        ("  clear_attributes(keys) -> Collection", None),
        ("    Example: collection.clear_attributes(['owner_team'])", None),
        ("", None),
        ("  delete_collection() -> None", None),
        ("    Example: collection.delete_collection()", None),
        ("    Note: Deletes collection and all associated data (permanent)", None),
        ("", None),
        ("  delete_all_items(group_name='data') -> ClearResult", None),
        ("    Example: result = collection.delete_all_items()", None),
        ("    Note: Clears items in group without deleting the collection", None),
        ("", None),
        ("Information:", "bold"),
        ("  print_info() -> None", None),
        ("    Example: collection.print_info()", None),
    ]

    if console:
        for text, style in help_text:
            if style == "bold":
                console.print(f"[bold]{text}[/bold]")
            elif style == "dim":
                console.print(f"[dim]{text}[/dim]")
            elif text.startswith("  ") and "()" in text:
                # Method signature - use accent color
                method_part = text.split(" -> ")[0]
                return_part = text.split(" -> ")[1] if " -> " in text else ""
                if return_part:
                    console.print(f"  [{accent_color}]{method_part}[/{accent_color}] -> {return_part}")
                else:
                    console.print(f"  [{accent_color}]{method_part}[/{accent_color}]")
            elif text.startswith("    Example:") or text.startswith("    Note:"):
                # Examples and notes - use muted color
                console.print(f"[{muted_color}]{text}[/{muted_color}]")
            else:
                console.print(text)
    else:
        # Plain text fallback
        if file is None:
            file = sys.stdout
        for text, _ in help_text:
            print(text, file=file)

print_nexus_info

print_nexus_info(logger_name: str, default_timeout: float, runner_status: str, client_status: str, base_url: Optional[str] = None, verify_ssl: Optional[bool] = None, projects: Optional[list[dict[str, Any]]] = None, file: Optional[TextIO] = None) -> None

Print comprehensive information about the Nexus client and projects.

Parameters:

Name Type Description Default
logger_name str

Logger name

required
default_timeout float

Default timeout in seconds

required
runner_status str

Runner status ("Active" or "Inactive")

required
client_status str

Client status ("Connected" or "Disconnected")

required
base_url Optional[str]

Optional base URL

None
verify_ssl Optional[bool]

Optional SSL verification setting

None
projects Optional[list[dict[str, Any]]]

Optional list of project dictionaries

None
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_nexus_info(
    logger_name: str,
    default_timeout: float,
    runner_status: str,
    client_status: str,
    base_url: Optional[str] = None,
    verify_ssl: Optional[bool] = None,
    projects: Optional[list[dict[str, Any]]] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print comprehensive information about the Nexus client and projects.

    Args:
        logger_name: Logger name
        default_timeout: Default timeout in seconds
        runner_status: Runner status ("Active" or "Inactive")
        client_status: Client status ("Connected" or "Disconnected")
        base_url: Optional base URL
        verify_ssl: Optional SSL verification setting
        projects: Optional list of project dictionaries
        file: Optional file to write to (defaults to stdout)
    """
    import os

    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")
    error_color = get_rich_style("error")

    if console:
        console.print("[bold]NEXUS CLIENT INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print()
        console.print("[bold]Client Configuration:[/bold]")
        console.print(f"   Logger Name: [{muted_color}]{logger_name}[/{muted_color}]")
        console.print(f"   Default Timeout: [{muted_color}]{default_timeout}s[/{muted_color}]")
        console.print(f"   Runner Status: [{primary_color}]{runner_status}[/{primary_color}]")
        console.print(f"   Client Status: [{primary_color}]{client_status}[/{primary_color}]")

        # Environment info
        console.print()
        console.print("[bold]Environment Variables:[/bold]")
        key_env_vars = ["MIURA_NEXUS_API_KEY", "MIURA_BASE_URL"]
        for var in key_env_vars:
            value = os.environ.get(var, "Not set")
            if var == "MIURA_NEXUS_API_KEY" and value != "Not set":
                value = f"{value[:10]}...{value[-10:]}" if len(value) > 20 else "***"
            console.print(f"   {var}: [{muted_color}]{value}[/{muted_color}]")

        # Base URL info
        if base_url is not None:
            console.print()
            console.print("[bold]Connection Details:[/bold]")
            console.print(f"   Base URL: [{muted_color}]{base_url}[/{muted_color}]")
            if verify_ssl is not None:
                console.print(f"   SSL Verification: [{muted_color}]{verify_ssl}[/{muted_color}]")

        # Projects information
        console.print()
        console.print("[bold]PROJECTS INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print()

        if projects is None:
            console.print(f"[{error_color}]Error: Could not list projects[/{error_color}]")
        elif not projects:
            console.print(f"[{muted_color}]No projects found[/{muted_color}]")
        else:
            console.print(f"Found {len(projects)} project{'s' if len(projects) != 1 else ''}:")
            console.print("[dim]" + "-" * 60 + "[/dim]")
            console.print()

            for i, project in enumerate(projects, 1):
                project_name = project.get("name", "Unnamed")
                project_uuid = project.get("uuid", "N/A")
                metadata = _normalized_metadata(project.get("metadata", {}))
                owner_profile = project.get("owner_profile")
                if owner_profile is None:
                    owner_profile = normalize_owner_profile(project.get("owner"))
                owner_line = format_project_owner_line(
                    str(project.get("owner_id", "") or ""),
                    owner_profile=owner_profile if isinstance(owner_profile, dict) else None,
                    metadata=metadata,
                )

                console.print(f"{i}. [{primary_color}]{project_name}[/{primary_color}]")
                console.print(f"   UUID: [{muted_color}]{project_uuid}[/{muted_color}]")
                if owner_line:
                    console.print(f"   Owner: [{muted_color}]{owner_line}[/{muted_color}]")

                # Handle metadata
                created_at, updated_at = _display_timestamps(
                    project.get("created_at"), project.get("updated_at"), metadata
                )
                metadata_display = _metadata_for_display(metadata)
                if metadata_display:
                    console.print("   Metadata:")
                    _print_metadata_value(
                        metadata_display,
                        indent="     ",
                        console=console,
                        primary_color=primary_color,
                        muted_color=muted_color,
                    )

                # Handle timestamps
                if created_at:
                    console.print(f"   Created: [{muted_color}]{created_at}[/{muted_color}]")
                if updated_at:
                    console.print(f"   Updated: [{muted_color}]{updated_at}[/{muted_color}]")

                console.print("[dim]" + "-" * 40 + "[/dim]")
                console.print()
    else:
        # Plain text fallback
        if file is None:
            file = sys.stdout
        print("NEXUS CLIENT INFORMATION", file=file)
        print("=" * 60, file=file)
        print("Client Configuration:", file=file)
        print(f"   Logger Name: {logger_name}", file=file)
        print(f"   Default Timeout: {default_timeout}s", file=file)
        print(f"   Runner Status: {runner_status}", file=file)
        print(f"   Client Status: {client_status}", file=file)
        print("\nEnvironment Variables:", file=file)
        key_env_vars = ["MIURA_NEXUS_API_KEY", "MIURA_BASE_URL"]
        for var in key_env_vars:
            value = os.environ.get(var, "Not set")
            if var == "MIURA_NEXUS_API_KEY" and value != "Not set":
                value = f"{value[:10]}...{value[-10:]}" if len(value) > 20 else "***"
            print(f"   {var}: {value}", file=file)
        if base_url is not None:
            print("\nConnection Details:", file=file)
            print(f"   Base URL: {base_url}", file=file)
            if verify_ssl is not None:
                print(f"   SSL Verification: {verify_ssl}", file=file)
        print("\nPROJECTS INFORMATION", file=file)
        print("=" * 60, file=file)
        if projects is None:
            print("Error: Could not list projects", file=file)
        elif not projects:
            print("No projects found", file=file)
        else:
            print(f"Found {len(projects)} project(s):", file=file)
            print("-" * 60, file=file)
            for i, project in enumerate(projects, 1):
                print(f"{i}. {project.get('name', 'Unnamed')}", file=file)
                print(f"   UUID: {project.get('uuid', 'N/A')}", file=file)
                metadata = _normalized_metadata(project.get("metadata", {}))
                owner_profile = project.get("owner_profile")
                if owner_profile is None:
                    owner_profile = normalize_owner_profile(project.get("owner"))
                owner_line = format_project_owner_line(
                    str(project.get("owner_id", "") or ""),
                    owner_profile=owner_profile if isinstance(owner_profile, dict) else None,
                    metadata=metadata,
                )
                if owner_line:
                    print(f"   Owner: {owner_line}", file=file)
                created_at, updated_at = _display_timestamps(
                    project.get("created_at"), project.get("updated_at"), metadata
                )
                metadata_display = _metadata_for_display(metadata)
                if metadata_display:
                    print("   Metadata:", file=file)
                    _print_metadata_value(metadata_display, indent="     ", file=file)
                if created_at:
                    print(f"   Created: {created_at}", file=file)
                if updated_at:
                    print(f"   Updated: {updated_at}", file=file)
                print("-" * 40, file=file)

print_nexus_help

print_nexus_help(file: Optional[TextIO] = None) -> None

Print help information for Nexus API methods.

Parameters:

Name Type Description Default
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_nexus_help(file: Optional[TextIO] = None) -> None:
    """
    Print help information for Nexus API methods.

    Args:
        file: Optional file to write to (defaults to stdout)
    """
    console = create_console(file=file)
    muted_color = get_rich_style("muted")
    accent_color = get_rich_style("accent")

    help_text = [
        ("Nexus API - Available Methods", "bold"),
        ("=" * 60, "dim"),
        ("", None),
        ("Project Management:", "bold"),
        ("  list_projects() -> list[ProjectInfo]", None),
        ("    Example: projects = nexus.list_projects()", None),
        ("    Note: Returns ProjectInfo dataclass objects (use project.name, project.id)", None),
        ("", None),
        ("  get_project(project_name: str) -> Project", None),
        ("    Example: project = nexus.get_project('my-project')", None),
        ("    Note: Returns Project wrapper with .name, .id attributes", None),
        ("", None),
        ("  create_project(project_name: str) -> Project", None),
        ("    Example: project = nexus.create_project('my-project')", None),
        ("    Note: Creates new project and returns Project wrapper", None),
        ("", None),
        ("  delete_project(project_name: str) -> None", None),
        ("    Example: nexus.delete_project('my-project')", None),
        ("    Note: Deletes project and all associated resources (collections, data, etc.)", None),
        ("", None),
        ("  iter_projects(prefetch_pages=1, page_size=50) -> Iterator[ProjectInfo]", None),
        ("    Example: for project in nexus.iter_projects(): ...", None),
        ("    Note: Yields ProjectInfo dataclass objects (use project.name, project.id)", None),
        ("", None),
        ("Path Navigation:", "bold"),
        ("  get(path: str) -> Project | Collection", None),
        ("    Example: obj = nexus.get('project/collection')", None),
        ("", None),
        ("  download(path: str, local_path: str, confirm=True) -> dict", None),
        ("    Example: result = nexus.download('project/collection', './downloads/')", None),
        ("", None),
        ("Resource Management:", "bold"),
        ("  close() -> None", None),
        ("    Example: nexus.close()", None),
        ("", None),
        ("  with nexus: ...  # Context manager", None),
        ("    Example: with Nexus() as nexus: ...", None),
    ]

    if console:
        for text, style in help_text:
            if style == "bold":
                console.print(f"[bold]{text}[/bold]")
            elif style == "dim":
                console.print(f"[dim]{text}[/dim]")
            elif text.startswith("  ") and "()" in text:
                method_part = text.split(" -> ")[0]
                return_part = text.split(" -> ")[1] if " -> " in text else ""
                if return_part:
                    console.print(f"  [{accent_color}]{method_part}[/{accent_color}] -> {return_part}")
                else:
                    console.print(f"  [{accent_color}]{method_part}[/{accent_color}]")
            elif text.startswith("    Example:") or text.startswith("    Note:"):
                console.print(f"[{muted_color}]{text}[/{muted_color}]")
            else:
                console.print(text)
    else:
        if file is None:
            file = sys.stdout
        for text, _ in help_text:
            print(text, file=file)

print_datasource_info

print_datasource_info(path: str, file_count: int, size_bytes: int, sample_files: Optional[list[str]] = None, file: Optional[TextIO] = None) -> None

Print information about a data source.

Parameters:

Name Type Description Default
path str

Data source path

required
file_count int

Number of files

required
size_bytes int

Total size in bytes

required
sample_files Optional[list[str]]

Optional list of sample file paths (first 5 will be shown)

None
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_datasource_info(
    path: str,
    file_count: int,
    size_bytes: int,
    sample_files: Optional[list[str]] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print information about a data source.

    Args:
        path: Data source path
        file_count: Number of files
        size_bytes: Total size in bytes
        sample_files: Optional list of sample file paths (first 5 will be shown)
        file: Optional file to write to (defaults to stdout)
    """
    from miura.display._formatting import human_bytes

    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")

    if console:
        console.print("[bold]LOCAL DATA SOURCE INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print(f"Path: [{primary_color}]{path}[/{primary_color}]")
        console.print(f"Files: [{muted_color}]{file_count}[/{muted_color}]")
        console.print(f"Size: [{muted_color}]{human_bytes(size_bytes)}[/{muted_color}]")

        if sample_files:
            console.print()
            console.print("[bold]Sample files:[/bold]")
            for i, file_path in enumerate(sample_files[:5], 1):
                console.print(f"   {i}. [{primary_color}]{file_path}[/{primary_color}]")
            if len(sample_files) > 5:
                console.print(f"   [{muted_color}]... and {len(sample_files) - 5} more files[/{muted_color}]")
        else:
            console.print()
            console.print(f"[{muted_color}](no files found)[/{muted_color}]")
    else:
        if file is None:
            file = sys.stdout
        print("LOCAL DATA SOURCE INFORMATION", file=file)
        print("=" * 60, file=file)
        print(f"Path: {path}", file=file)
        print(f"Files: {file_count}", file=file)
        print(f"Size: {human_bytes(size_bytes)}", file=file)
        if sample_files:
            print("\nSample files:", file=file)
            for i, file_path in enumerate(sample_files[:5], 1):
                print(f"   {i + 1}. {file_path}", file=file)
            if len(sample_files) > 5:
                print(f"   ... and {len(sample_files) - 5} more files", file=file)
        else:
            print("\n(no files found)", file=file)

print_recipe_info

print_recipe_info(recipe: RecipeResponse, recent_runs: Optional[list[RunHandle]] = None, file: Optional[TextIO] = None) -> None

Print comprehensive information about a recipe.

Parameters:

Name Type Description Default
recipe RecipeResponse

RecipeResponse object with recipe details

required
recent_runs Optional[list[RunHandle]]

Optional recent runs for this recipe (already filtered/limited)

None
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_recipe_info(
    recipe: "RecipeResponse",
    recent_runs: Optional[list["RunHandle"]] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print comprehensive information about a recipe.

    Args:
        recipe: RecipeResponse object with recipe details
        recent_runs: Optional recent runs for this recipe (already filtered/limited)
        file: Optional file to write to (defaults to stdout)
    """
    # Lazy import to avoid circular dependency
    from miura.nexus.recipe.recipe_models import RecipeResponse  # noqa: F401

    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")

    if console:
        console.print("[bold]RECIPE INFORMATION[/bold]")
        console.print("[dim]" + "=" * 60 + "[/dim]")
        console.print()
        console.print("[bold]Recipe Details:[/bold]")
        console.print(f"   Name: [{primary_color}]{recipe.name}[/{primary_color}]")
        console.print(f"   UUID: [{muted_color}]{recipe.id}[/{muted_color}]")
        console.print(f"   Project ID: [{muted_color}]{recipe.project_id}[/{muted_color}]")
        if recipe.description:
            console.print(f"   Description: [{muted_color}]{recipe.description}[/{muted_color}]")
        if recipe.created_by:
            console.print(f"   Created by: [{muted_color}]{recipe.created_by}[/{muted_color}]")

        # Sources information
        console.print()
        console.print("[bold]Sources:[/bold]")
        if recipe.sources:
            for i, source in enumerate(recipe.sources, 1):
                console.print(
                    f"   {i}. [{primary_color}]{source.collection_name}[/{primary_color}] "
                    f"(group: [{muted_color}]{source.group_name}[/{muted_color}])"
                )
        else:
            console.print(f"   [{muted_color}]No sources (data integrity issue)[/{muted_color}]")

        # Targets information
        console.print()
        console.print("[bold]Targets:[/bold]")
        if recipe.targets:
            for i, target in enumerate(recipe.targets, 1):
                console.print(
                    f"   {i}. [{primary_color}]{target.collection_name}[/{primary_color}] "
                    f"(group: [{muted_color}]{target.group_name}[/{muted_color}])"
                )
        else:
            console.print(f"   [{muted_color}]Omitted (will be created during execution)[/{muted_color}]")

        # Recipe configuration
        console.print()
        console.print("[bold]Recipe Configuration:[/bold]")
        if recipe.recipe:
            _print_metadata_value(
                recipe.recipe,
                indent="   ",
                console=console,
                primary_color=primary_color,
                muted_color=muted_color,
            )
        else:
            console.print(f"   [{muted_color}]No configuration[/{muted_color}]")

        # Recent runs
        console.print()
        console.print("[bold]Recent Runs:[/bold]")
        if recent_runs is None:
            console.print(f"   [{muted_color}]Run history not requested[/{muted_color}]")
        elif not recent_runs:
            console.print(f"   [{muted_color}]No runs found for this recipe[/{muted_color}]")
        else:
            for i, run in enumerate(recent_runs, 1):
                state_str = getattr(run.state, "value", str(run.state))
                created_at = str(run.created_at) if run.created_at is not None else "-"
                ended_at = str(run.ended_at) if run.ended_at is not None else "-"
                trigger_status = str(run.trigger_status) if run.trigger_status is not None else "-"
                console.print(
                    f"   {i}. [{primary_color}]{run.id}[/{primary_color}] "
                    f"[{muted_color}]state={state_str}, created={created_at}, ended={ended_at}, "
                    f"trigger={trigger_status}[/{muted_color}]"
                )
    else:
        # Plain text fallback
        if file is None:
            file = sys.stdout
        print("RECIPE INFORMATION", file=file)
        print("=" * 60, file=file)
        print("Recipe Details:", file=file)
        print(f"   Name: {recipe.name}", file=file)
        print(f"   UUID: {recipe.id}", file=file)
        print(f"   Project ID: {recipe.project_id}", file=file)
        if recipe.description:
            print(f"   Description: {recipe.description}", file=file)
        if recipe.created_by:
            print(f"   Created by: {recipe.created_by}", file=file)
        print("\nSources:", file=file)
        if recipe.sources:
            for i, source in enumerate(recipe.sources, 1):
                print(f"   {i}. {source.collection_name} (group: {source.group_name})", file=file)
        else:
            print("   No sources (data integrity issue)", file=file)
        print("\nTargets:", file=file)
        if recipe.targets:
            for i, target in enumerate(recipe.targets, 1):
                print(f"   {i}. {target.collection_name} (group: {target.group_name})", file=file)
        else:
            print("   Omitted (will be created during execution)", file=file)
        print("\nRecipe Configuration:", file=file)
        if recipe.recipe:
            _print_metadata_value(recipe.recipe, indent="   ", file=file)
        else:
            print("   No configuration", file=file)

        print("\nRecent Runs:", file=file)
        if recent_runs is None:
            print("   Run history not requested", file=file)
        elif not recent_runs:
            print("   No runs found for this recipe", file=file)
        else:
            for i, run in enumerate(recent_runs, 1):
                state_str = getattr(run.state, "value", str(run.state))
                created_at = str(run.created_at) if run.created_at is not None else "-"
                ended_at = str(run.ended_at) if run.ended_at is not None else "-"
                trigger_status = str(run.trigger_status) if run.trigger_status is not None else "-"
                print(
                    f"   {i}. {run.id} state={state_str}, created={created_at}, ended={ended_at}, "
                    f"trigger={trigger_status}",
                    file=file,
                )

print_capabilities

print_capabilities(caps: Capabilities, reader_limit: int = 5, writer_limit: Optional[int] = None, file: Optional[TextIO] = None) -> None

Print recipe capabilities (readers, writers) from recipe-service..

Parameters:

Name Type Description Default
caps Capabilities

Capabilities from project.get_recipe_capabilities()

required
reader_limit int

Max reader ids to show in preview (default 5)

5
writer_limit Optional[int]

Max writer ids to show (default: all)

None
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_capabilities(
    caps: "Capabilities",
    reader_limit: int = 5,
    writer_limit: Optional[int] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print recipe capabilities (readers, writers) from recipe-service..

    Args:
        caps: Capabilities from project.get_recipe_capabilities()
        reader_limit: Max reader ids to show in preview (default 5)
        writer_limit: Max writer ids to show (default: all)
        file: Optional file to write to (defaults to stdout)
    """
    from miura.nexus.recipe.recipe_models import Capabilities
    from .formatters import _format_capabilities

    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")

    if console and isinstance(caps, Capabilities):
        console.print(f"  version: [{muted_color}]{caps.version}[/{muted_color}]")
        reader_ids = [r.id for r in caps.readers]
        writer_ids = [w.id for w in caps.writers]
        suffix_r = "..." if len(reader_ids) > reader_limit else ""
        console.print(
            f"  readers ({len(reader_ids)}): [{primary_color}]{reader_ids[:reader_limit]}{suffix_r}[/{primary_color}]"
        )
        w_limit = writer_limit if writer_limit is not None else len(writer_ids)
        suffix_w = "..." if len(writer_ids) > w_limit else ""
        console.print(
            f"  writers ({len(writer_ids)}): [{primary_color}]{writer_ids[:w_limit]}{suffix_w}[/{primary_color}]"
        )
    else:
        if file is None:
            file = sys.stdout
        print(_format_capabilities(caps, reader_limit=reader_limit, writer_limit=writer_limit), file=file)

print_validation_result

print_validation_result(validation: ValidationResult, max_issues: int = 10, file: Optional[TextIO] = None) -> None

Print recipe validation result (is_valid + issues). Option C.

Parameters:

Name Type Description Default
validation ValidationResult

ValidationResult from project.validate_recipe()

required
max_issues int

Max issues to list (default 10)

10
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_validation_result(
    validation: "ValidationResult",
    max_issues: int = 10,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print recipe validation result (is_valid + issues). Option C.

    Args:
        validation: ValidationResult from project.validate_recipe()
        max_issues: Max issues to list (default 10)
        file: Optional file to write to (defaults to stdout)
    """
    from miura.nexus.recipe.recipe_models import ValidationResult
    from .formatters import _format_validation_result

    console = create_console(file=file)
    error_color = get_rich_style("error")
    success_color = get_rich_style("success")
    muted_color = get_rich_style("muted")

    if console and isinstance(validation, ValidationResult):
        valid_style = success_color if validation.is_valid else error_color
        console.print(f"  is_valid: [{valid_style}]{validation.is_valid}[/{valid_style}]")
        console.print(f"  issues: [{muted_color}]{len(validation.issues)}[/{muted_color}]")
        for issue in validation.issues[:max_issues]:
            path_part = f" [{muted_color}](path={issue.path})[/{muted_color}]" if issue.path else ""
            console.print(f"  - [{error_color}][{issue.code}][/{error_color}] {issue.message}{path_part}")
        if len(validation.issues) > max_issues:
            console.print(f"  [{muted_color}]... and {len(validation.issues) - max_issues} more[/{muted_color}]")
    else:
        if file is None:
            file = sys.stdout
        print(_format_validation_result(validation, max_issues=max_issues), file=file)

print_invalid_recipe

print_invalid_recipe(invalid: InvalidRecipe, max_issues: int = 10, file: Optional[TextIO] = None) -> None

Print InvalidRecipe outcome (validation failed, no create).

Parameters:

Name Type Description Default
invalid InvalidRecipe

InvalidRecipe from create_recipe / builder.create()

required
max_issues int

Max issues to list (default 10)

10
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_invalid_recipe(
    invalid: "InvalidRecipe",
    max_issues: int = 10,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print InvalidRecipe outcome (validation failed, no create).

    Args:
        invalid: InvalidRecipe from create_recipe / builder.create()
        max_issues: Max issues to list (default 10)
        file: Optional file to write to (defaults to stdout)
    """
    from miura.nexus.recipe.recipe_models import InvalidRecipe
    from .formatters import _format_invalid_recipe

    console = create_console(file=file)
    error_color = get_rich_style("error")
    muted_color = get_rich_style("muted")

    if console and isinstance(invalid, InvalidRecipe):
        v = invalid.validation
        console.print(f"  [{error_color}]InvalidRecipe: issues={len(v.issues)}[/{error_color}]")
        for issue in v.issues[:max_issues]:
            path_part = f" [{muted_color}](path={issue.path})[/{muted_color}]" if issue.path else ""
            console.print(f"  - [{error_color}][{issue.code}][/{error_color}] {issue.message}{path_part}")
        if len(v.issues) > max_issues:
            console.print(f"  [{muted_color}]... and {len(v.issues) - max_issues} more[/{muted_color}]")
    else:
        if file is None:
            file = sys.stdout
        print(_format_invalid_recipe(invalid, max_issues=max_issues), file=file)

print_created_recipe

print_created_recipe(recipe: RecipeResponse, run: Optional[RunHandle] = None, file: Optional[TextIO] = None) -> None

Print CreatedRecipe outcome (recipe created, optional run).

Parameters:

Name Type Description Default
recipe RecipeResponse

RecipeResponse from create

required
run Optional[RunHandle]

Optional RunHandle if submit=True

None
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_created_recipe(
    recipe: "RecipeResponse",
    run: Optional["RunHandle"] = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print CreatedRecipe outcome (recipe created, optional run).

    Args:
        recipe: RecipeResponse from create
        run: Optional RunHandle if submit=True
        file: Optional file to write to (defaults to stdout)
    """
    from miura.nexus.recipe.recipe_models import RecipeResponse
    from .formatters import _format_created_recipe

    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")

    if console and isinstance(recipe, RecipeResponse):
        run_str = f", run={run}" if run is not None else ", run=None"
        console.print(
            f"  CreatedRecipe: recipe=[{primary_color}]{recipe.name}[/{primary_color}] "
            f"[{muted_color}]({recipe.id})[/{muted_color}]{run_str}"
        )
    else:
        if file is None:
            file = sys.stdout
        print(_format_created_recipe(recipe, run=run), file=file)

print_recipe_create_result

print_recipe_create_result(result: InvalidRecipe | CreatedRecipe, max_issues: int = 10, file: Optional[TextIO] = None) -> None

Print create_recipe / builder.create() result (InvalidRecipe or CreatedRecipe).

Parameters:

Name Type Description Default
result InvalidRecipe | CreatedRecipe

InvalidRecipe or CreatedRecipe

required
max_issues int

Max validation issues to show when InvalidRecipe (default 10)

10
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_recipe_create_result(
    result: "InvalidRecipe | CreatedRecipe",
    max_issues: int = 10,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print create_recipe / builder.create() result (InvalidRecipe or CreatedRecipe).

    Args:
        result: InvalidRecipe or CreatedRecipe
        max_issues: Max validation issues to show when InvalidRecipe (default 10)
        file: Optional file to write to (defaults to stdout)
    """
    from miura.nexus.recipe.recipe_models import InvalidRecipe, CreatedRecipe

    if isinstance(result, InvalidRecipe):
        print_invalid_recipe(result, max_issues=max_issues, file=file)
    elif isinstance(result, CreatedRecipe):
        print_created_recipe(result.recipe, run=result.run, file=file)
    else:
        if file is None:
            file = sys.stdout
        print(f"  {result!r}", file=file)

print_run_triggered

print_run_triggered(run: RunHandle, file: Optional[TextIO] = None) -> None

Print the trigger response for a recipe run in a user-friendly format.

Use this when a run has just been started (e.g. after run_recipe() or execute_recipe()) to show run id, state, and pertinent metadata.

Parameters:

Name Type Description Default
run RunHandle

RunHandle from the trigger/execute response (id, state, and optional metadata)

required
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_run_triggered(run: "RunHandle", file: Optional[TextIO] = None) -> None:
    """
    Print the trigger response for a recipe run in a user-friendly format.

    Use this when a run has just been started (e.g. after run_recipe() or
    execute_recipe()) to show run id, state, and pertinent metadata.

    Args:
        run: RunHandle from the trigger/execute response (id, state, and optional metadata)
        file: Optional file to write to (defaults to stdout)
    """
    from miura.nexus.recipe.recipe_execution import RunHandle  # noqa: F401

    state_str = getattr(run.state, "value", str(run.state))
    short_id = run.id[:8] + "…" if len(run.id) > 8 else run.id

    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")
    success_color = get_rich_style("success")

    # RunHandle may have execute-response fields (trigger_status, dag_run_id, triggered_at)
    # and/or monitoring fields (recipe_id, project_id, etc.); show what is present
    trigger_status = getattr(run, "trigger_status", None)
    triggered_at = getattr(run, "triggered_at", None)
    dag_run_id = getattr(run, "dag_run_id", None)
    trigger_error_code = getattr(run, "trigger_error_code", None)

    if console:
        console.print("[bold]Run triggered[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(
            f"   Run ID:   [{primary_color}]{short_id}[/{primary_color}]  [{muted_color}]{run.id}[/{muted_color}]"
        )
        console.print(f"   State:    [{success_color}]{state_str}[/{success_color}]")
        if trigger_status is not None:
            console.print(f"   Trigger:  [{muted_color}]{trigger_status}[/{muted_color}]")
        if triggered_at is not None:
            console.print(f"   Triggered at: [{muted_color}]{triggered_at}[/{muted_color}]")
        if dag_run_id is not None:
            console.print(f"   DAG run:  [{muted_color}]{dag_run_id}[/{muted_color}]")
        if trigger_error_code is not None:
            console.print(f"   Trigger error: [{muted_color}]{trigger_error_code}[/{muted_color}]")
        if run.created_at:
            console.print(f"   Created:  [{muted_color}]{run.created_at}[/{muted_color}]")
        if run.recipe_id:
            console.print(f"   Recipe:   [{muted_color}]{run.recipe_id}[/{muted_color}]")
        if run.project_id:
            console.print(f"   Project:  [{muted_color}]{run.project_id}[/{muted_color}]")
        if run.triggered_by:
            console.print(f"   Triggered by: [{muted_color}]{run.triggered_by}[/{muted_color}]")
        if run.tags:
            console.print(f"   Tags:     [{muted_color}]{run.tags}[/{muted_color}]")
        console.print()
    else:
        if file is None:
            file = sys.stdout
        print("Run triggered", file=file)
        print("─" * 50, file=file)
        print(f"   Run ID:   {short_id}  {run.id}", file=file)
        print(f"   State:    {state_str}", file=file)
        if trigger_status is not None:
            print(f"   Trigger:  {trigger_status}", file=file)
        if triggered_at is not None:
            print(f"   Triggered at: {triggered_at}", file=file)
        if dag_run_id is not None:
            print(f"   DAG run:  {dag_run_id}", file=file)
        if trigger_error_code is not None:
            print(f"   Trigger error: {trigger_error_code}", file=file)
        if run.created_at:
            print(f"   Created:  {run.created_at}", file=file)
        if run.recipe_id:
            print(f"   Recipe:   {run.recipe_id}", file=file)
        if run.project_id:
            print(f"   Project:  {run.project_id}", file=file)
        if run.triggered_by:
            print(f"   Triggered by: {run.triggered_by}", file=file)
        if run.tags:
            print(f"   Tags:     {run.tags}", file=file)
        print(file=file)

print_run_info

print_run_info(run: RunHandle, file: Optional[TextIO] = None) -> None

Print a run snapshot in a user-friendly format.

This is intended for run list/detail views where a run may already contain monitoring fields and optional diagnostics.

Parameters:

Name Type Description Default
run RunHandle

RunHandle with list/monitoring fields

required
file Optional[TextIO]

Optional file to write to (defaults to stdout)

None
Source code in src/miura/display/info.py
def print_run_info(run: "RunHandle", file: Optional[TextIO] = None) -> None:
    """
    Print a run snapshot in a user-friendly format.

    This is intended for run list/detail views where a run may already contain
    monitoring fields and optional diagnostics.

    Args:
        run: RunHandle with list/monitoring fields
        file: Optional file to write to (defaults to stdout)
    """
    from miura.nexus.recipe.recipe_execution import RunHandle  # noqa: F401

    state_str = getattr(run.state, "value", str(run.state))
    short_id = run.id[:8] + "…" if len(run.id) > 8 else run.id

    trigger_status = getattr(run, "trigger_status", None)
    triggered_at = getattr(run, "triggered_at", None)
    dag_run_id = getattr(run, "dag_run_id", None)
    trigger_error_code = getattr(run, "trigger_error_code", None)
    execution_error = getattr(run, "execution_error", None)
    diagnostics = getattr(run, "diagnostics", None)

    console = create_console(file=file)
    primary_color = get_rich_style("primary")
    muted_color = get_rich_style("muted")
    success_color = get_rich_style("success")
    error_color = get_rich_style("error")

    if console:
        console.print("[bold]RUN INFORMATION[/bold]")
        console.print("[dim]" + "─" * 60 + "[/dim]")
        console.print(
            f"   Run ID:      [{primary_color}]{short_id}[/{primary_color}] " f"[{muted_color}]{run.id}[/{muted_color}]"
        )
        console.print(f"   State:       [{success_color}]{state_str}[/{success_color}]")
        if run.project_id:
            console.print(f"   Project:     [{muted_color}]{run.project_id}[/{muted_color}]")
        if run.recipe_id:
            console.print(f"   Recipe:      [{muted_color}]{run.recipe_id}[/{muted_color}]")
        if run.created_at:
            console.print(f"   Created:     [{muted_color}]{run.created_at}[/{muted_color}]")
        if run.updated_at:
            console.print(f"   Updated:     [{muted_color}]{run.updated_at}[/{muted_color}]")
        if run.started_at:
            console.print(f"   Started:     [{muted_color}]{run.started_at}[/{muted_color}]")
        if run.ended_at:
            console.print(f"   Ended:       [{muted_color}]{run.ended_at}[/{muted_color}]")
        if run.triggered_by:
            console.print(f"   Triggered by:[{muted_color}] {run.triggered_by}[/{muted_color}]")
        if trigger_status is not None:
            console.print(f"   Trigger:     [{muted_color}]{trigger_status}[/{muted_color}]")
        if triggered_at is not None:
            console.print(f"   Triggered at:[{muted_color}] {triggered_at}[/{muted_color}]")
        if dag_run_id is not None:
            console.print(f"   DAG run:     [{muted_color}]{dag_run_id}[/{muted_color}]")
        if trigger_error_code is not None:
            console.print(f"   Trigger error:[{error_color}] {trigger_error_code}[/{error_color}]")
        if execution_error is not None:
            console.print(f"   Execution error:[{error_color}] {execution_error}[/{error_color}]")
        if run.tags:
            console.print(f"   Tags:        [{muted_color}]{run.tags}[/{muted_color}]")
        if getattr(run, "code_ref", None):
            console.print("   Code ref:")
            _print_metadata_value(
                run.code_ref,
                indent="     ",
                console=console,
                primary_color=primary_color,
                muted_color=muted_color,
            )
        if getattr(run, "params", None):
            console.print("   Params:")
            _print_metadata_value(
                run.params,
                indent="     ",
                console=console,
                primary_color=primary_color,
                muted_color=muted_color,
            )
        if diagnostics is not None:
            if diagnostics.has_error and diagnostics.error is not None:
                detail = f" detail={diagnostics.error.detail}" if diagnostics.error.detail else ""
                console.print(
                    f"   Diagnostics: [{error_color}]error={diagnostics.error.code.value}{detail}[/{error_color}]"
                )
            else:
                console.print(f"   Diagnostics: [{muted_color}]tasks={len(diagnostics.tasks)}[/{muted_color}]")
        console.print()
    else:
        if file is None:
            file = sys.stdout
        print("RUN INFORMATION", file=file)
        print("─" * 60, file=file)
        print(f"   Run ID:      {short_id}  {run.id}", file=file)
        print(f"   State:       {state_str}", file=file)
        if run.project_id:
            print(f"   Project:     {run.project_id}", file=file)
        if run.recipe_id:
            print(f"   Recipe:      {run.recipe_id}", file=file)
        if run.created_at:
            print(f"   Created:     {run.created_at}", file=file)
        if run.updated_at:
            print(f"   Updated:     {run.updated_at}", file=file)
        if run.started_at:
            print(f"   Started:     {run.started_at}", file=file)
        if run.ended_at:
            print(f"   Ended:       {run.ended_at}", file=file)
        if run.triggered_by:
            print(f"   Triggered by:{run.triggered_by}", file=file)
        if trigger_status is not None:
            print(f"   Trigger:     {trigger_status}", file=file)
        if triggered_at is not None:
            print(f"   Triggered at:{triggered_at}", file=file)
        if dag_run_id is not None:
            print(f"   DAG run:     {dag_run_id}", file=file)
        if trigger_error_code is not None:
            print(f"   Trigger error: {trigger_error_code}", file=file)
        if execution_error is not None:
            print(f"   Execution error: {execution_error}", file=file)
        if run.tags:
            print(f"   Tags:        {run.tags}", file=file)
        if getattr(run, "code_ref", None):
            print("   Code ref:", file=file)
            _print_metadata_value(run.code_ref, indent="     ", file=file)
        if getattr(run, "params", None):
            print("   Params:", file=file)
            _print_metadata_value(run.params, indent="     ", file=file)
        if diagnostics is not None:
            if diagnostics.has_error and diagnostics.error is not None:
                detail = f" detail={diagnostics.error.detail}" if diagnostics.error.detail else ""
                print(f"   Diagnostics: error={diagnostics.error.code.value}{detail}", file=file)
            else:
                print(f"   Diagnostics: tasks={len(diagnostics.tasks)}", file=file)
        print(file=file)

format_run_status_line

format_run_status_line(run_id: str, state: Any, elapsed: float, *, pending_timeout_remaining: Optional[float] = None, prefix: Optional[str] = None) -> str

Format a single run-status line: state label + elapsed (e.g. "Pending... 10s", "Success! total 44s"). Spinner is added by Progress as first character for the live line. run_id, prefix, and pending_timeout_remaining are accepted for API compatibility but not shown in the simple format.

Source code in src/miura/display/status_line.py
def format_run_status_line(
    run_id: str,
    state: Any,
    elapsed: float,
    *,
    pending_timeout_remaining: Optional[float] = None,
    prefix: Optional[str] = None,
) -> str:
    """
    Format a single run-status line: state label + elapsed (e.g. "Pending... 10s",
    "Success! total 44s"). Spinner is added by Progress as first character for
    the live line. run_id, prefix, and pending_timeout_remaining are accepted
    for API compatibility but not shown in the simple format.
    """
    del run_id, pending_timeout_remaining, prefix  # unused in simple format
    state_str = getattr(state, "value", str(state))
    label = _STATE_LABELS.get(state_str.upper() if isinstance(state_str, str) else state_str, str(state_str))
    if state_str in ("SUCCESS", "FAILED", "CANCELED"):
        return f"{label} total {elapsed:.0f}s"
    return f"{label} {elapsed:.0f}s"

print_json

print_json(obj: Any, *, title: Optional[str] = None, indent: int = 2, sort_keys: bool = True, file: Optional[TextIO] = None) -> None

Pretty-print an object as JSON. Handles dataclass, pydantic, and plain dict/list.

Parameters:

Name Type Description Default
obj Any

Object to serialize (dict, dataclass, or pydantic model).

required
title Optional[str]

Optional line printed before the JSON.

None
indent int

JSON indent (default 2).

2
sort_keys bool

Sort dict keys (default True).

True
file Optional[TextIO]

Output stream (default stdout).

None
Source code in src/miura/display/dump.py
def print_json(
    obj: Any,
    *,
    title: Optional[str] = None,
    indent: int = 2,
    sort_keys: bool = True,
    file: Optional[TextIO] = None,
) -> None:
    """
    Pretty-print an object as JSON. Handles dataclass, pydantic, and plain dict/list.

    Args:
        obj: Object to serialize (dict, dataclass, or pydantic model).
        title: Optional line printed before the JSON.
        indent: JSON indent (default 2).
        sort_keys: Sort dict keys (default True).
        file: Output stream (default stdout).
    """
    if file is None:
        file = sys.stdout
    if title is not None:
        print(title, file=file)
    data = _to_builtin(obj)
    print(json.dumps(data, indent=indent, sort_keys=sort_keys), file=file)

print_download_progress

print_download_progress(*, completed: int, total_files: int, failed: int = 0, bytes_received: int = 0, bytes_total: int | None = None, elapsed_seconds: float, file: Optional[TextIO] = None) -> None

Print periodic download progress to stdout (mirrors upload progress UX).

Source code in src/miura/display/download.py
def print_download_progress(
    *,
    completed: int,
    total_files: int,
    failed: int = 0,
    bytes_received: int = 0,
    bytes_total: int | None = None,
    elapsed_seconds: float,
    file: Optional[TextIO] = None,
) -> None:
    """Print periodic download progress to stdout (mirrors upload progress UX)."""
    status = format_transfer_progress(
        verb="Downloading",
        completed=completed,
        total_files=total_files,
        failed=failed,
        bytes_current=bytes_received,
        bytes_total=bytes_total,
        elapsed_seconds=elapsed_seconds,
    )
    out = file or sys.stdout
    print(status, file=out)

print_download_result

print_download_result(download_result: Any, *, max_errors: int = 10, file: Optional[TextIO] = None) -> None

Print :class:~miura.sdk.types.DownloadResult or download-service dict.

Source code in src/miura/display/download.py
def print_download_result(
    download_result: Any,
    *,
    max_errors: int = 10,
    file: Optional[TextIO] = None,
) -> None:
    """Print :class:`~miura.sdk.types.DownloadResult` or download-service dict."""
    if isinstance(download_result, dict):
        status = download_result.get("status", "unknown")
        ok = int(download_result.get("files_downloaded", 0) or 0)
        failed = int(download_result.get("files_failed", 0) or 0)
        total_bytes = int(download_result.get("downloaded_size", 0) or 0)
        destination = str(download_result.get("destination", "") or "")
        errors = download_result.get("errors") or []
    else:
        status = getattr(download_result, "status", "unknown")
        ok = int(getattr(download_result, "successful_count", 0) or 0)
        failed = int(getattr(download_result, "failed_count", 0) or 0)
        total_bytes = int(getattr(download_result, "total_size_bytes", 0) or 0)
        destination = str(getattr(download_result, "destination", "") or "")
        errors = getattr(download_result, "errors", None) or []

    console = create_console(file=file)
    muted = get_rich_style("muted")
    success = get_rich_style("success")
    error = get_rich_style("error")
    status_style = success if status == "completed" and failed == 0 else error

    if console:
        console.print("[bold]Download result[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(f"   Status:    [{status_style}]{status}[/{status_style}]")
        if destination:
            console.print(f"   Destination: [{muted}]{destination}[/{muted}]")
        console.print(
            f"   Downloaded: [{muted}]{ok}[/{muted}]  "
            f"Failed: [{error if failed else muted}]{failed}[/{error if failed else muted}]  "
            f"Size: [{muted}]{human_bytes(total_bytes)}[/{muted}]"
        )
        if errors:
            console.print(f"   Errors:    [{error}]{len(errors)}[/{error}]")
            for err in errors[:max_errors]:
                console.print(f"   • [{error}]{err}[/{error}]")
            if len(errors) > max_errors:
                console.print(f"   [{muted}]… and {len(errors) - max_errors} more[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print("Download result", file=out)
        print("─" * 50, file=out)
        print(f"   Status:    {status}", file=out)
        if destination:
            print(f"   Destination: {destination}", file=out)
        print(
            f"   Downloaded: {ok}  failed={failed}  size={human_bytes(total_bytes)}",
            file=out,
        )
        if errors:
            print(f"   Errors ({len(errors)}):", file=out)
            for err in errors[:max_errors]:
                print(f"   • {err}", file=out)
            if len(errors) > max_errors:
                print(f"   … and {len(errors) - max_errors} more", file=out)
        print(file=out)

format_duration_compact

format_duration_compact(seconds: float) -> str

Format duration for progress lines (e.g. 45s, 1m25s, 1h12m).

Source code in src/miura/display/_formatting.py
def format_duration_compact(seconds: float) -> str:
    """Format duration for progress lines (e.g. 45s, 1m25s, 1h12m)."""
    s = max(0, int(seconds))
    if s < 60:
        return f"{s}s"
    mins, secs = divmod(s, 60)
    if mins < 60:
        return f"{mins}m{secs}s" if secs else f"{mins}m"
    hours, rem_mins = divmod(mins, 60)
    if rem_mins:
        return f"{hours}h{rem_mins}m"
    return f"{hours}h"

format_log_snapshot_line

format_log_snapshot_line(primary_text: str, elapsed_s: float, *, run_id_short: str | None = None, is_terminal: bool = False) -> str

One-line recipe progress log: state-first text + compact duration.

Source code in src/miura/display/progress_formatters.py
def format_log_snapshot_line(
    primary_text: str,
    elapsed_s: float,
    *,
    run_id_short: str | None = None,
    is_terminal: bool = False,
) -> str:
    """One-line recipe progress log: state-first text + compact duration."""
    duration = format_duration_compact(elapsed_s)
    if primary_text.endswith(duration) or f"{_SEPARATOR}{duration}" in primary_text:
        return primary_text
    duration_part = f"total {duration}" if is_terminal else duration
    line = f"{primary_text}{_SEPARATOR}{duration_part}"
    if run_id_short:
        return f"{line}{_SEPARATOR}run {run_id_short}"
    return line

format_transfer_completion

format_transfer_completion(*, successful: int, failed: int = 0, elapsed_seconds: float, verbose: bool = False, successful_singlepart: int = 0, successful_multipart: int = 0) -> str

State-first upload completion summary.

Source code in src/miura/display/progress_formatters.py
def format_transfer_completion(
    *,
    successful: int,
    failed: int = 0,
    elapsed_seconds: float,
    verbose: bool = False,
    successful_singlepart: int = 0,
    successful_multipart: int = 0,
) -> str:
    """State-first upload completion summary."""
    duration = format_duration_compact(elapsed_seconds)
    if verbose:
        return (
            f"Success · {successful} files uploaded "
            f"({successful_singlepart} singlepart, {successful_multipart} multipart) · "
            f"{failed} failed · {duration}"
        )
    if failed == 0:
        return f"Success · {successful} files uploaded · {duration}"
    total = successful + failed
    return f"Partial success · {successful}/{total} files uploaded · {failed} failed · {duration}"

format_transfer_progress

format_transfer_progress(*, verb: str, completed: int, total_files: int, failed: int = 0, bytes_current: int = 0, bytes_total: int | None = None, elapsed_seconds: float) -> str

State-first upload/download progress line.

Source code in src/miura/display/progress_formatters.py
def format_transfer_progress(
    *,
    verb: str,
    completed: int,
    total_files: int,
    failed: int = 0,
    bytes_current: int = 0,
    bytes_total: int | None = None,
    elapsed_seconds: float,
) -> str:
    """State-first upload/download progress line."""
    parts = [verb, f"{completed}/{total_files} files complete"]
    if failed > 0:
        parts.append(f"{failed} failed")
    if bytes_total is not None:
        parts.append(f"{human_bytes(bytes_current)} / {human_bytes(bytes_total)}")
    parts.append(format_duration_compact(elapsed_seconds))
    return _SEPARATOR.join(parts)

print_batch_stream_stats

print_batch_stream_stats(stats: Any, *, label: str = 'Catalog batch stream', file: Optional[TextIO] = None) -> None

Print observability counters for a :class:~miura.adapters.torch.BatchStreamStats.

Source code in src/miura/display/catalog.py
def print_batch_stream_stats(
    stats: Any,
    *,
    label: str = "Catalog batch stream",
    file: Optional[TextIO] = None,
) -> None:
    """Print observability counters for a :class:`~miura.adapters.torch.BatchStreamStats`."""
    console = create_console(file=file)
    muted = get_rich_style("muted")
    success = get_rich_style("success")

    rows = [
        ("stream_iterations", stats.stream_iterations),
        ("stream_batches", stats.stream_batches),
        ("stream_rows", stats.stream_rows),
        ("stream_bytes", human_bytes(stats.stream_bytes) if stats.stream_bytes else "0 B"),
        ("torch_steps", stats.torch_steps),
        ("consumer_stopped_early", stats.consumer_stopped_early),
        ("producer_duration_s", f"{stats.producer_duration_s:.4f}"),
        ("consumer_duration_s", f"{stats.consumer_duration_s:.4f}"),
        ("producer_queue_put_wait_s", f"{stats.producer_queue_put_wait_s:.4f}"),
        ("consumer_queue_get_wait_s", f"{stats.consumer_queue_get_wait_s:.4f}"),
    ]

    if console:
        console.print(f"[bold]{label}[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        for key, value in rows:
            console.print(f"   {key:<28} [{muted}]{value}[/{muted}]")
        console.print(f"[{success}]✓[/{success}] batch stream stats")
    else:
        out = file or sys.stdout
        print(label, file=out)
        print("─" * 50, file=out)
        for key, value in rows:
            print(f"   {key}: {value}", file=out)
        print("✓ batch stream stats", file=out)

print_catalog_read_batches

print_catalog_read_batches(*, batch_count: int, row_count: int, column_names: Sequence[str] | None = None, sample_batch: RecordBatch | None = None, file: Optional[TextIO] = None) -> None

Summarize consumed Arrow batches.

Source code in src/miura/display/catalog.py
def print_catalog_read_batches(
    *,
    batch_count: int,
    row_count: int,
    column_names: Sequence[str] | None = None,
    sample_batch: pa.RecordBatch | None = None,
    file: Optional[TextIO] = None,
) -> None:
    """Summarize consumed Arrow batches."""
    console = create_console(file=file)
    primary = get_rich_style("primary")
    muted = get_rich_style("muted")
    success = get_rich_style("success")

    nbytes = 0
    if sample_batch is not None:
        try:
            nbytes = sample_batch.nbytes
        except Exception:
            nbytes = 0

    if console:
        console.print("[bold]Catalog read batches[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(
            f"   Batches:   [{primary}]{batch_count}[/{primary}]  " f"Rows: [{success}]{row_count}[/{success}]"
        )
        if column_names:
            console.print(f"   Columns:   [{muted}]{', '.join(column_names)}[/{muted}]")
        if nbytes:
            console.print(f"   1st batch: [{muted}]{human_bytes(nbytes)}[/{muted}] Arrow in-memory")
        console.print()
    else:
        out = file or sys.stdout
        print("Catalog read batches", file=out)
        print("─" * 50, file=out)
        print(f"   Batches: {batch_count}  Rows: {row_count}", file=out)
        if column_names:
            print(f"   Columns: {', '.join(column_names)}", file=out)
        if nbytes:
            print(f"   1st batch: {human_bytes(nbytes)} Arrow in-memory", file=out)
        print(file=out)

print_catalog_read_complete

print_catalog_read_complete(*, label: str, batch_count: int, row_count: int, file: Optional[TextIO] = None) -> None

Final success line for catalog read examples.

Source code in src/miura/display/catalog.py
def print_catalog_read_complete(
    *,
    label: str,
    batch_count: int,
    row_count: int,
    file: Optional[TextIO] = None,
) -> None:
    """Final success line for catalog read examples."""
    console = create_console(file=file)
    success = get_rich_style("success")

    if console:
        console.print(f"[{success}]✓[/{success}] [bold]{label}[/bold] " f"— batches={batch_count} rows={row_count}")
    else:
        out = file or sys.stdout
        print(f"✓ {label} — batches={batch_count} rows={row_count}", file=out)

print_catalog_read_ineligible

print_catalog_read_ineligible(*, asset_id: str, error: BaseException, file: Optional[TextIO] = None) -> None

Print expected ineligibility failure (409 ConflictError).

Source code in src/miura/display/catalog.py
def print_catalog_read_ineligible(
    *,
    asset_id: str,
    error: BaseException,
    file: Optional[TextIO] = None,
) -> None:
    """Print expected ineligibility failure (409 ConflictError)."""
    console = create_console(file=file)
    warning = get_rich_style("warning")
    muted = get_rich_style("muted")

    if console:
        console.print("[bold]Catalog read rejected (expected)[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(f"   Asset:     [{muted}]{asset_id}[/{muted}]")
        console.print(f"   Error:     [{warning}]{type(error).__name__}[/{warning}] {error}")
        console.print()
        console.print(f"[{warning}]✓[/{warning}] ineligible asset demo — fail-closed as expected")
    else:
        out = file or sys.stdout
        print("Catalog read rejected (expected)", file=out)
        print("─" * 50, file=out)
        print(f"   Asset:  {asset_id}", file=out)
        print(f"   Error:  {type(error).__name__}: {error}", file=out)
        print("✓ ineligible asset demo — fail-closed as expected", file=out)

print_catalog_read_run

print_catalog_read_run(run: Any, *, batch_size: int | None = None, shuffle: dict[str, Any] | None = None, file: Optional[TextIO] = None) -> None

Print CatalogReadRun identifiers and run options.

Source code in src/miura/display/catalog.py
def print_catalog_read_run(
    run: Any,
    *,
    batch_size: int | None = None,
    shuffle: dict[str, Any] | None = None,
    file: Optional[TextIO] = None,
) -> None:
    """Print ``CatalogReadRun`` identifiers and run options."""
    console = create_console(file=file)
    primary = get_rich_style("primary")
    muted = get_rich_style("muted")
    run_id = getattr(run, "run_id", "")

    if console:
        console.print("[bold]Catalog read run[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        _id_line("Run:", run_id, primary=primary, muted=muted, console=console)
        if batch_size is not None:
            console.print(f"   Batch size [{muted}]{batch_size}[/{muted}]")
        if shuffle is not None:
            console.print(f"   Shuffle:   [{muted}]{shuffle}[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print("Catalog read run", file=out)
        print("─" * 50, file=out)
        print(f"   Run:        {run_id}", file=out)
        if batch_size is not None:
            print(f"   Batch size: {batch_size}", file=out)
        if shuffle is not None:
            print(f"   Shuffle:    {shuffle}", file=out)
        print(file=out)

print_catalog_read_scope

print_catalog_read_scope(*, project_id: str, collection_id: str, asset_id: str, tenant_id: str = '', columns: Sequence[str] | None = None, title: str = 'Catalog read scope', file: Optional[TextIO] = None) -> None

Print explicit catalog scope.

Source code in src/miura/display/catalog.py
def print_catalog_read_scope(
    *,
    project_id: str,
    collection_id: str,
    asset_id: str,
    tenant_id: str = "",
    columns: Sequence[str] | None = None,
    title: str = "Catalog read scope",
    file: Optional[TextIO] = None,
) -> None:
    """Print explicit catalog scope."""
    console = create_console(file=file)
    primary = get_rich_style("primary")
    muted = get_rich_style("muted")

    if console:
        console.print(f"[bold]{title}[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        if tenant_id:
            console.print(f"   Tenant:    [{muted}]{tenant_id}[/{muted}]")
        _id_line("Project:", project_id, primary=primary, muted=muted, console=console)
        console.print(f"   Collection: [{primary}]{collection_id}[/{primary}]")
        _id_line("Asset:", asset_id, primary=primary, muted=muted, console=console)
        if columns:
            console.print(f"   Columns:   [{muted}]{', '.join(columns)}[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print(title, file=out)
        print("─" * 50, file=out)
        if tenant_id:
            print(f"   Tenant:     {tenant_id}", file=out)
        print(f"   Project:    {project_id}", file=out)
        print(f"   Collection: {collection_id}", file=out)
        print(f"   Asset:      {asset_id}", file=out)
        if columns:
            print(f"   Columns:    {', '.join(columns)}", file=out)
        print(file=out)

print_catalog_read_session

print_catalog_read_session(session: Any, *, file: Optional[TextIO] = None) -> None

Print CatalogReadSession identifiers.

Source code in src/miura/display/catalog.py
def print_catalog_read_session(
    session: Any,
    *,
    file: Optional[TextIO] = None,
) -> None:
    """Print ``CatalogReadSession`` identifiers."""
    console = create_console(file=file)
    primary = get_rich_style("primary")
    muted = get_rich_style("muted")
    session_id = getattr(session, "session_id", "")

    if console:
        console.print("[bold]Catalog read session[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        _id_line("Session:", session_id, primary=primary, muted=muted, console=console)
        console.print(f"   Asset:     [{muted}]{getattr(session, 'asset_id', '')}[/{muted}]")
        console.print(f"   Collection: [{muted}]{getattr(session, 'collection_id', '')}[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print("Catalog read session", file=out)
        print("─" * 50, file=out)
        print(f"   Session:    {session_id}", file=out)
        print(f"   Asset:      {getattr(session, 'asset_id', '')}", file=out)
        print(f"   Collection: {getattr(session, 'collection_id', '')}", file=out)
        print(file=out)

format_ingestion_plan_gate_failure

format_ingestion_plan_gate_failure(plan: _IngestionPlanLike, *, report: PlanValidationReport | None = None, max_issues: int = 5) -> str

One-line message when the plan gate blocks apply.

Source code in src/miura/display/ingestion.py
def format_ingestion_plan_gate_failure(
    plan: _IngestionPlanLike,
    *,
    report: PlanValidationReport | None = None,
    max_issues: int = 5,
) -> str:
    """One-line message when the plan gate blocks apply."""
    short = ellipsize_identifier(plan.ingestion_plan_id, head=10, tail=8)
    gate_status = _plan_gate_status(plan)
    base = (
        f"Plan gate failed: plan_id={short} "
        f"blocking={plan.blocking_failure_count} gate={gate_status} state={plan.state}"
    )
    if report is None or not report.failed_records:
        return base
    lines = [format_validation_issue(issue) for issue in report.failed_records[:max_issues]]
    suffix = ""
    if len(report.failed_records) > max_issues:
        suffix = f" (+{len(report.failed_records) - max_issues} more)"
    return base + " — " + "; ".join(lines) + suffix

format_local_upload_apply_failure

format_local_upload_apply_failure(upload_result: Any, *, max_errors: int = 5) -> str

One-line message when :meth:IngestionPlan.apply did not succeed.

Source code in src/miura/display/ingestion.py
def format_local_upload_apply_failure(upload_result: Any, *, max_errors: int = 5) -> str:
    """One-line message when :meth:`IngestionPlan.apply` did not succeed."""
    errors = getattr(upload_result, "validation_errors", None) or getattr(upload_result, "errors", None) or []
    status = getattr(upload_result, "status", "unknown")
    run_id = getattr(upload_result, "ingestion_run_id", None) or ""
    plan_id = getattr(upload_result, "ingestion_plan_id", None) or ""
    session_id = getattr(upload_result, "upload_id", None) or ""
    id_bits = []
    if plan_id:
        id_bits.append(f"plan={ellipsize_identifier(plan_id, head=8, tail=6)}")
    if run_id:
        id_bits.append(f"run={ellipsize_identifier(run_id, head=8, tail=6)}")
    if session_id:
        id_bits.append(f"session={ellipsize_identifier(session_id, head=8, tail=6)}")
    id_suffix = f" ({', '.join(id_bits)})" if id_bits else ""
    if not errors:
        return f"plan.apply failed: status={status}{id_suffix}"
    shown = errors[:max_errors]
    suffix = f" (+{len(errors) - max_errors} more)" if len(errors) > max_errors else ""
    return f"plan.apply failed: status={status}{id_suffix} errors={shown}{suffix}"

format_validation_issue

format_validation_issue(issue: ValidationIssue) -> str

Canonical formatter for a plan gate :class:~miura.sdk.types.ValidationIssue.

Source code in src/miura/display/ingestion.py
def format_validation_issue(issue: ValidationIssue) -> str:
    """Canonical formatter for a plan gate :class:`~miura.sdk.types.ValidationIssue`."""
    code = issue.code or "error"
    if issue.path and issue.message:
        return f"[{code}] {issue.path}: {issue.message}"
    if issue.object_key and issue.message:
        return f"[{code}] {issue.object_key}: {issue.message}"
    if issue.message:
        return f"[{code}] {issue.message}"
    return str(code)

format_validation_record

format_validation_record(record: ValidationIssue) -> str

Alias for :func:format_validation_issue.

Source code in src/miura/display/ingestion.py
def format_validation_record(record: ValidationIssue) -> str:
    """Alias for :func:`format_validation_issue`."""
    return format_validation_issue(record)

gate_failure_message

gate_failure_message(plan: _IngestionPlanLike, *, report: PlanValidationReport | None = None, max_issues: int = 5) -> str

Programmatic gate failure text for exceptions and SystemExit.

Source code in src/miura/display/ingestion.py
def gate_failure_message(
    plan: _IngestionPlanLike,
    *,
    report: PlanValidationReport | None = None,
    max_issues: int = 5,
) -> str:
    """Programmatic gate failure text for exceptions and ``SystemExit``."""
    return format_ingestion_plan_gate_failure(plan, report=report, max_issues=max_issues)

print_files

print_files(body: dict[str, Any], *, max_items: int = 20, file: Optional[TextIO] = None) -> None

Print a paginated collection files listing (raw LAS JSON items / files payload).

For product browse use :func:miura.display.print_items with iter_items instead.

Parameters:

Name Type Description Default
body dict[str, Any]

Raw response dict (internal/debug; not from a public collection API).

required
max_items int

Max paths to show.

20
Source code in src/miura/display/ingestion.py
def print_files(
    body: dict[str, Any],
    *,
    max_items: int = 20,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print a paginated collection **files** listing (raw LAS JSON ``items`` / ``files`` payload).

    For product browse use :func:`miura.display.print_items` with ``iter_items`` instead.

    Args:
        body: Raw response dict (internal/debug; not from a public collection API).
        max_items: Max paths to show.
    """
    items = body.get("items") or body.get("files") or []
    if not isinstance(items, list):
        items = []
    total = body.get("total") or body.get("count") or len(items)
    n = len(items)
    count_msg = f"Found {n} item{'s' if n != 1 else ''}" if n else "Found 0 items"
    if total and int(total) != n:
        count_msg = f"{count_msg} (total≈{total})"

    console = create_console(file=file)
    muted = get_rich_style("muted")
    primary = get_rich_style("primary")

    if console:
        console.print(f"[bold]{count_msg}[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        for row in items[:max_items]:
            path = _file_entry_path(row)
            if path:
                console.print(f"   • [{primary}]{path}[/{primary}]")
        if len(items) > max_items:
            console.print(f"   [{muted}]… and {len(items) - max_items} more[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print(count_msg, file=out)
        print("─" * 50, file=out)
        for row in items[:max_items]:
            path = _file_entry_path(row)
            if path:
                print(f"   • {path}", file=out)
        if len(items) > max_items:
            print(f"   … and {len(items) - max_items} more", file=out)
        print(file=out)

print_ingestion_plan

print_ingestion_plan(plan: _IngestionPlanLike, *, data_root: str | Path | None = None, file: Optional[TextIO] = None) -> None

Print ingestion plan gate summary (dry-run result).

Accepts :class:~miura.nexus.ingestion.IngestionPlan or :class:~miura.sdk.types.LocalUploadPlan.

Source code in src/miura/display/ingestion.py
def print_ingestion_plan(
    plan: _IngestionPlanLike,
    *,
    data_root: str | Path | None = None,
    file: Optional[TextIO] = None,
) -> None:
    """
    Print ingestion plan gate summary (dry-run result).

    Accepts :class:`~miura.nexus.ingestion.IngestionPlan` or
    :class:`~miura.sdk.types.LocalUploadPlan`.
    """
    root = _resolve_data_root(plan, data_root)
    short_id = ellipsize_identifier(plan.ingestion_plan_id, head=10, tail=8)
    admissible = bool(getattr(plan, "is_admissible", False))
    gate_status = _plan_gate_status(plan)
    total_size_bytes = _plan_total_size_bytes(plan)
    run_type = str(getattr(plan, "run_type", "") or "")

    console = create_console(file=file)
    primary = get_rich_style("primary")
    muted = get_rich_style("muted")
    success = get_rich_style("success")
    error = get_rich_style("error")
    gate_style = success if admissible else error

    if console:
        console.print("[bold]Ingestion plan[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(f"   Plan ID:  [{primary}]{short_id}[/{primary}] " f"[{muted}]{plan.ingestion_plan_id}[/{muted}]")
        mode = getattr(plan, "collection_mode", None)
        if mode:
            console.print(f"   Mode:     [{muted}]{mode}[/{muted}]")
        if run_type:
            console.print(f"   Type:     [{muted}]{run_type}[/{muted}]")
        console.print(f"   Gate:     [{gate_style}]{gate_status}[/{gate_style}]")
        console.print(f"   State:    [{muted}]{plan.state}[/{muted}]")
        console.print(f"   Admissible: [{gate_style}]{admissible}[/{gate_style}]")
        candidates = f"   Candidates: [{muted}]{plan.object_count}[/{muted}]"
        if total_size_bytes is not None:
            candidates += f"  Size: [{muted}]{human_bytes(total_size_bytes)}[/{muted}]"
        console.print(candidates)
        for line in _work_summary_lines(plan):
            console.print(f"   [{muted}]{line}[/{muted}]")
        if plan.blocking_failure_count:
            console.print(f"   Blocking: [{error}]{plan.blocking_failure_count}[/{error}]")
        if root:
            console.print(f"   Data root: [{muted}]{root}[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print("Ingestion plan", file=out)
        print("─" * 50, file=out)
        print(f"   Plan ID:  {short_id}  {plan.ingestion_plan_id}", file=out)
        mode = getattr(plan, "collection_mode", None)
        if mode:
            print(f"   Mode:     {mode}", file=out)
        if run_type:
            print(f"   Type:     {run_type}", file=out)
        print(f"   Gate:     {gate_status}  admissible={admissible}", file=out)
        print(f"   State:    {plan.state}", file=out)
        candidates = f"   Candidates: {plan.object_count}"
        if total_size_bytes is not None:
            candidates += f"  size={human_bytes(total_size_bytes)}"
        print(candidates, file=out)
        for line in _work_summary_lines(plan):
            print(f"   {line}", file=out)
        if plan.blocking_failure_count:
            print(f"   Blocking: {plan.blocking_failure_count}", file=out)
        if root:
            print(f"   Data root: {root}", file=out)
        print(file=out)

print_local_upload_complete

print_local_upload_complete(plan: _IngestionPlanLike, upload_result: Any, *, data_root: str | Path | None = None, file: Optional[TextIO] = None) -> None

Print success summary after plan.apply (ingestion session + counts).

Source code in src/miura/display/ingestion.py
def print_local_upload_complete(
    plan: _IngestionPlanLike,
    upload_result: Any,
    *,
    data_root: str | Path | None = None,
    file: Optional[TextIO] = None,
) -> None:
    """Print success summary after plan.apply (ingestion session + counts)."""
    root = _resolve_data_root(plan, data_root)
    plan_short = ellipsize_identifier(plan.ingestion_plan_id, head=10, tail=8)
    session_id = getattr(upload_result, "upload_id", None) or ""
    session_short = ellipsize_identifier(session_id, head=10, tail=8) if session_id else "—"
    status = getattr(upload_result, "status", "unknown")
    ok = getattr(upload_result, "successful_count", None)
    if ok is None:
        ok = getattr(upload_result, "files_uploaded", plan.object_count)
    noop = ok == 0 and status == "success" and not getattr(plan, "requires_execution", True)
    files_label = f"{ok} (already in collection)" if noop else f"{ok} (planned {plan.object_count})"

    console = create_console(file=file)
    primary = get_rich_style("primary")
    muted = get_rich_style("muted")
    success = get_rich_style("success")

    if console:
        console.print("[bold]LOCAL_UPLOAD complete[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(f"   Status:   [{success}]{status}[/{success}]")
        console.print(
            f"   Plan ID:  [{primary}]{plan_short}[/{primary}] " f"[{muted}]{plan.ingestion_plan_id}[/{muted}]"
        )
        if session_id:
            console.print(f"   Session:  [{primary}]{session_short}[/{primary}] " f"[{muted}]{session_id}[/{muted}]")
        console.print(f"   Files:    [{muted}]{files_label}[/{muted}]")
        if root:
            console.print(f"   Data root: [{muted}]{root}[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print("LOCAL_UPLOAD complete", file=out)
        print("─" * 50, file=out)
        print(f"   Status:   {status}", file=out)
        print(f"   Plan ID:  {plan_short}  {plan.ingestion_plan_id}", file=out)
        if session_id:
            print(f"   Session:  {session_short}  {session_id}", file=out)
        print(f"   Files:    {files_label}", file=out)
        if root:
            print(f"   Data root: {root}", file=out)
        print(file=out)

print_plan_validation_report

print_plan_validation_report(report: PlanValidationReport, *, max_issues: int = 20, max_passed: int = 5, status: str = 'failed', file: Optional[TextIO] = None) -> None

Print typed plan gate diagnostics from a :class:~miura.sdk.types.PlanValidationReport.

Source code in src/miura/display/ingestion.py
def print_plan_validation_report(
    report: PlanValidationReport,
    *,
    max_issues: int = 20,
    max_passed: int = 5,
    status: str = "failed",
    file: Optional[TextIO] = None,
) -> None:
    """Print typed plan gate diagnostics from a :class:`~miura.sdk.types.PlanValidationReport`."""
    issues = report.failed_records if status == "failed" else report.records
    passed = report.passed_records if status == "failed" else ()
    console = create_console(file=file)
    error = get_rich_style("error")
    muted = get_rich_style("muted")
    primary = get_rich_style("primary")
    success = get_rich_style("success")
    short = ellipsize_identifier(report.ingestion_plan_id, head=10, tail=8)
    title = "Plan validation report (failed)" if status == "failed" else "Plan validation report"

    if console:
        console.print(f"[bold]{title}[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(f"   Plan ID: [{primary}]{short}[/{primary}] " f"[{muted}]{report.ingestion_plan_id}[/{muted}]")
        if report.manifest_sha256:
            console.print(f"   Manifest: [{muted}]{report.manifest_sha256[:16]}…[/{muted}]")
        if not issues and not passed:
            console.print(f"   [{muted}]No matching issues.[/{muted}]")
        if issues:
            console.print(f"   [{error}]Failed ({len(report.failed_records)}):[/{error}]")
            for issue in issues[:max_issues]:
                console.print(f"   • [{error}]{format_validation_issue(issue)}[/{error}]")
            if len(issues) > max_issues:
                console.print(f"   [{muted}]… and {len(issues) - max_issues} more[/{muted}]")
        if passed:
            console.print(f"   [{success}]Passed ({len(passed)}):[/{success}]")
            for issue in passed[:max_passed]:
                console.print(f"   • [{success}]{format_passed_record(issue)}[/{success}]")
            if len(passed) > max_passed:
                console.print(f"   [{muted}]… and {len(passed) - max_passed} more[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print(title, file=out)
        print("─" * 50, file=out)
        print(f"   Plan ID: {short}  {report.ingestion_plan_id}", file=out)
        if issues:
            print(f"   Failed ({len(report.failed_records)}):", file=out)
            for issue in issues[:max_issues]:
                print(f"   • {format_validation_issue(issue)}", file=out)
            if len(issues) > max_issues:
                print(f"   … and {len(issues) - max_issues} more", file=out)
        if passed:
            print(f"   Passed ({len(passed)}):", file=out)
            for issue in passed[:max_passed]:
                print(f"   • {format_passed_record(issue)}", file=out)
            if len(passed) > max_passed:
                print(f"   … and {len(passed) - max_passed} more", file=out)
        print(file=out)

print_upload_result

print_upload_result(upload_result: Any, *, max_errors: int = 10, file: Optional[TextIO] = None) -> None

Print :class:~miura.sdk.types.UploadResult from plan.apply.

Source code in src/miura/display/ingestion.py
def print_upload_result(
    upload_result: Any,
    *,
    max_errors: int = 10,
    file: Optional[TextIO] = None,
) -> None:
    """Print :class:`~miura.sdk.types.UploadResult` from ``plan.apply``."""
    status = getattr(upload_result, "status", "unknown")
    ok = int(getattr(upload_result, "successful_count", 0) or getattr(upload_result, "files_uploaded", 0) or 0)
    failed = int(getattr(upload_result, "failed_count", 0) or getattr(upload_result, "files_failed", 0) or 0)
    total_bytes = _result_total_size_bytes(upload_result)
    session_id = getattr(upload_result, "upload_id", None) or ""
    run_id = getattr(upload_result, "ingestion_run_id", None) or ""
    plan_id = getattr(upload_result, "ingestion_plan_id", None) or ""
    errors = getattr(upload_result, "validation_errors", None) or getattr(upload_result, "errors", None) or []

    console = create_console(file=file)
    primary = get_rich_style("primary")
    muted = get_rich_style("muted")
    success = get_rich_style("success")
    error = get_rich_style("error")
    status_style = success if status == "success" else error

    if console:
        console.print("[bold]Upload result[/bold]")
        console.print("[dim]" + "─" * 50 + "[/dim]")
        console.print(f"   Status:    [{status_style}]{status}[/{status_style}]")
        if session_id:
            short = ellipsize_identifier(session_id, head=10, tail=8)
            console.print(f"   Session:   [{primary}]{short}[/{primary}] " f"[{muted}]{session_id}[/{muted}]")
        if run_id:
            short = ellipsize_identifier(run_id, head=10, tail=8)
            console.print(f"   Run:       [{primary}]{short}[/{primary}] " f"[{muted}]{run_id}[/{muted}]")
        if plan_id:
            short = ellipsize_identifier(plan_id, head=10, tail=8)
            console.print(f"   Plan:      [{primary}]{short}[/{primary}] " f"[{muted}]{plan_id}[/{muted}]")
        counts = (
            f"   Uploaded:  [{muted}]{ok}[/{muted}]  "
            f"Failed: [{error if failed else muted}]{failed}[/{error if failed else muted}]"
        )
        if total_bytes is not None:
            counts += f"  Size: [{muted}]{human_bytes(total_bytes)}[/{muted}]"
        console.print(counts)
        if errors:
            console.print(f"   Errors:    [{error}]{len(errors)}[/{error}]")
            for err in errors[:max_errors]:
                console.print(f"   • [{error}]{_format_validation_error(err)}[/{error}]")
            if len(errors) > max_errors:
                console.print(f"   [{muted}]… and {len(errors) - max_errors} more[/{muted}]")
        console.print()
    else:
        out = file or sys.stdout
        print("Upload result", file=out)
        print("─" * 50, file=out)
        print(f"   Status:    {status}", file=out)
        if session_id:
            print(f"   Session:   {session_id}", file=out)
        if run_id:
            print(f"   Run:       {run_id}", file=out)
        if plan_id:
            print(f"   Plan:      {plan_id}", file=out)
        counts = f"   Uploaded:  {ok}  failed={failed}"
        if total_bytes is not None:
            counts += f"  size={human_bytes(total_bytes)}"
        print(counts, file=out)
        if errors:
            print(f"   Errors ({len(errors)}):", file=out)
            for err in errors[:max_errors]:
                print(f"   • {_format_validation_error(err)}", file=out)
            if len(errors) > max_errors:
                print(f"   … and {len(errors) - max_errors} more", file=out)
        print(file=out)