Skip to main content
Development utilities make local side effects explicit: commands return results, parallel jobs return task records, errors are formatted, and proxy settings are scoped.
Use this page when a script or extension needs operational helpers around a HeavenBase workflow.

​
1. Core Idea

These helpers are for the work around the product surface: running a command, fanning out local tasks, formatting a useful error, temporarily changing proxy environment variables, or redirecting logs while a demo runs. The common thread is reversibility and visibility. A helper should make side effects easier to inspect and easier to undo, not hide them behind a magical wrapper.

​
2. Run a Command

Use cmd(...) when you need stdout, stderr, and the exit code as data. 
from heavenbase.utils import cmd

result = cmd(["python", "--version"])

if not result.ok:
    raise RuntimeError(result.err)

print(result.out)
Ask for one field when the caller only needs that field: 
from heavenbase.utils import cmd

version = cmd(["python", "--version"], include="out")
handle = cmd(["python", "-m", "http.server", "8000"], wait=False, include="handle")
Sequence commands do not invoke a shell unless you pass shell=True. Prefer sequence commands for user-provided values.

​
3. Map Bounded Work

Use pmap(...) when work can run concurrently and the returned values must stay in input order. 
from heavenbase.utils import pmap

items = [{"score": 3}, {"score": 1}, {"score": 2}]
scores = pmap(lambda score: score * 2, items, max_workers=4)
Each item is passed as keyword arguments, so batch items must be mappings. Use batch(...) when you need per-item status, captured exceptions, elapsed time, or progress callbacks. 
from heavenbase.utils import batch

results = batch(lambda x: 10 / x, [{"x": 2}, {"x": 0}, {"x": 5}], max_workers=2)

failed = [result for result in results if not result.ok]
Async code uses the same contract through abatch(...) and abatch_stream(...).

​
4. Format Errors and Mismatches

Use raise_mismatch(...) when validation should suggest the closest supported value. 
from heavenbase.utils import capture_error, error_str, raise_mismatch

try:
    raise_mismatch(["json", "yaml"], "jsno", name="format")
except ValueError as exc:
    message = error_str(exc, tb=False)
    record = capture_error(exc)
capture_error(...) returns an ErrorRecord that can preserve the original exception through record.reraise().

​
5. Use Scoped Proxy Settings

Use NetworkProxy when one operation needs temporary HTTP_PROXY, HTTPS_PROXY, or NO_PROXY environment variables. 
from heavenbase.utils import NetworkProxy

with NetworkProxy(http_proxy="http://localhost:7890", no_proxy="localhost,127.0.0.1"):
    ...
Existing environment values are restored when the context exits. Passing an empty string disables that proxy variable inside the context.

​
6. Log and Color Diagnostics

Use get_logger(...) for HeavenBase log output, and redirect_logs(...) or suppress_logs(...) for temporary diagnostic routing. 
from heavenbase.utils import get_logger, redirect_logs, restore_logs, suppress_logs

logger = get_logger("heavenbase.docs")
logger.success("ready")

redirect_logs("demos/.temp/heavenbase.log")
try:
    logger.warning("written to the log file")
finally:
    restore_logs()

suppress_logs()
restore_logs()
Color helpers such as cstr(...), color_success(...), and strip_color(...) are for terminal diagnostics. Do not store colored strings in JSON, rows, or benchmark artifacts.

​
Further Exploration

Related resources:
  • File System - local paths and files used by command workflows.
  • Random - reproducible data for parallel tests and benchmarks.
  • Configuration - config defaults for parallel worker limits and logging.