live_print_scope

live_print_scope stores per-scope metadata in a contextvars ContextVar. print_to_live and log_to_live read the active scope: they prepend prefix to each line and skip output when suppress is true.

Use it when parallel work (for example tfdo multi-directory orchestration) needs tagged scroll lines without editing every print_to_live call site. Pair it with run_pool: submit copies the submitter's context into the worker thread. With run_and_wait, message_callbacks see the scope active when the run was created.

Read the active scope

from ask_shell.console import get_live_print_context, live_print_scope

with live_print_scope(prefix="[prod] "):
    ctx = get_live_print_context()
    print(repr(ctx.prefix if ctx else ""))
    #> '[prod] '

Suppress live output

from ask_shell.console import get_live_print_context, live_print_scope

with live_print_scope(suppress=True):
    ctx = get_live_print_context()
    print(ctx.suppress if ctx else False)
    #> True

Nested scopes restore the outer context when the inner block exits.

from ask_shell.console import get_live_print_context, live_print_scope

with live_print_scope(prefix="outer "):
    outer = get_live_print_context()
    with live_print_scope(prefix="inner "):
        inner = get_live_print_context()
    after = get_live_print_context()
print(f"{outer.prefix}|{inner.prefix}|{after.prefix}")
#> outer |inner |outer

Context at run_pool submit time

Set live_print_scope in the thread that calls submit. Each worker sees the scope that was active for its submit call.

from threading import Barrier

from ask_shell.console import get_live_print_context, live_print_scope
from ask_shell.shell import run_pool

results: list[str] = []
barrier = Barrier(2)


def worker() -> None:
    barrier.wait()
    ctx = get_live_print_context()
    results.append(ctx.prefix if ctx else "")


with run_pool("demo", total=2, pool_thread_count=2, max_concurrent_submits=2) as pool:
    with live_print_scope(prefix="a"):
        f_a = pool.submit(worker)
    with live_print_scope(prefix="b"):
        f_b = pool.submit(worker)
    f_a.result()
    f_b.result()

print(sorted(results))
#> ['a', 'b']

Context with run_and_wait and message_callbacks

ShellRun snapshots the active context when run / run_and_wait creates it. The queue consumer runs message_callbacks inside that snapshot, so handlers see the same live_print_scope as the caller even though dispatch runs on a pool thread.

from ask_shell._internal._run import run_and_wait
from ask_shell._internal.events import ShellRunStdOutput
from ask_shell.console import get_live_print_context, live_print_scope

seen: list[str | None] = []


def on_stdout(message):
    match message:
        case ShellRunStdOutput(is_stdout=True):
            ctx = get_live_print_context()
            seen.append(ctx.prefix if ctx else None)
    return False


with live_print_scope(prefix="[dir] "):
    run_and_wait("echo ok", message_callbacks=[on_stdout])

print(seen)
#> ['[dir] ']

The snapshot is fixed for the life of that ShellRun; changing live_print_scope after run_and_wait returns does not affect callbacks for an in-flight run. Retries reuse the same snapshot.