`io`#

Common I/O utilities.

Classes#

`DeltaSecondsFormatter`	Logging formatter with additional attributes for run time logging.
`ContextDecorator`	Base class for a context manager class (implementing __enter__() and __exit__()) that also
`SwallowBrokenPipe`	Ignore BrokenPipeError and errors related to stdout or stderr being closed by a downstream program.
`CaptureTarget`	Constants used for contextmanager captured.
`DummyExecutor`	Synchronous executor for debugging; executes tasks in the current thread.
`ThreadLimitedThreadPoolExecutor`	Thread pool executor that gracefully handles thread creation limits.
`time_recorder`	Context decorator for recording execution times to a CSV file.

Functions#

`dashlist`(→ str)	Format an iterable as a dashed list with optional indentation.
`env_vars`(→ collections.abc.Generator[None, None, None])	Temporarily set environment variables.
`env_var`(→ collections.abc.Generator[None, None, None])	Temporarily set a single environment variable.
`env_unmodified`(→ collections.abc.Generator[None, None, ...)	Context manager that ensures environment variables are not modified.
`captured`(→ collections.abc.Generator[Any, None, None])	Capture outputs of sys.stdout and sys.stderr.
`argv`(→ collections.abc.Generator[None, None, None])	Temporarily replace sys.argv with the given list.
`disable_logger`(→ collections.abc.Generator[None, None, ...)	Temporarily disable a logger by setting its level above CRITICAL.
`stderr_log_level`(→ collections.abc.Generator[None, ...)	Temporarily set a logger to output to stderr at the given level.
`attach_stderr_handler`(→ None)	Attach a new stderr handler to the given logger and configure both.
`timeout`(→ Any)	Enforce a maximum time for a callable to complete.
`get_instrumentation_record_file`(→ str)	Return the path to the instrumentation record file.
`print_instrumentation_data`(→ None)	Print aggregated instrumentation data from the record file as JSON.

Attributes#

`IS_INTERACTIVE`
`_FORMATTER`
`swallow_broken_pipe`
`as_completed`

IS_INTERACTIVE#

class DeltaSecondsFormatter(fmt: str | None = None, datefmt: str | None = None)#

Bases: logging.Formatter

Logging formatter with additional attributes for run time logging.

`delta_secs`: Elapsed seconds since last log/format call (or creation of logger).

`relative_created_secs`: Like relativeCreated, time relative to the initialization of the logging module but conveniently scaled to seconds as a float value.

Initialize the formatter with optional format strings.

Parameters:

fmt -- Log message format string.
datefmt -- Date format string.

prev_time#

format(record: logging.LogRecord) → str#

Format the specified record as text.

The record's attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

_FORMATTER#

dashlist(iterable: collections.abc.Iterable[Any], indent: int = 2) → str#

Format an iterable as a dashed list with optional indentation.

Parameters:

iterable -- Items to format.
indent -- Number of spaces for indentation.

Returns:

Formatted string with each item on a new line prefixed by ``- ``.

class ContextDecorator#

Base class for a context manager class (implementing __enter__() and __exit__()) that also makes it a decorator.

__call__(f: collections.abc.Callable[Ellipsis, Any]) → collections.abc.Callable[Ellipsis, Any]#

class SwallowBrokenPipe#

Bases: ContextDecorator

Ignore BrokenPipeError and errors related to stdout or stderr being closed by a downstream program.

__enter__() → None#

__exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) → bool | None#

swallow_broken_pipe#

class CaptureTarget#

Bases: enum.Enum

Constants used for contextmanager captured.

Used similarly like the constants PIPE, STDOUT for stdlib's subprocess.Popen.

STRING = -1#

STDOUT = -2#

env_vars(var_map: dict[str, str] | None = None, callback: collections.abc.Callable[[], None] | None = None, stack_callback: collections.abc.Callable[[bool], None] | None = None) → collections.abc.Generator[None, None, None]#

Temporarily set environment variables.

Parameters:

var_map -- Dictionary of environment variable names to values.
callback -- Optional callback invoked when entering and exiting the context.
stack_callback -- Optional callback invoked with True when entering, False when exiting.

env_var(name: str, value: str, callback: collections.abc.Callable[[], None] | None = None, stack_callback: collections.abc.Callable[[bool], None] | None = None) → collections.abc.Generator[None, None, None]#

Temporarily set a single environment variable.

Parameters:

name -- Environment variable name.
value -- Environment variable value.
callback -- Optional callback invoked when entering and exiting the context.
stack_callback -- Optional callback invoked with True when entering, False when exiting.

env_unmodified(callback: collections.abc.Callable[[], None] | None = None) → collections.abc.Generator[None, None, None]#

Context manager that ensures environment variables are not modified.

Parameters:: callback -- Optional callback invoked when entering and exiting the context.

captured(stdout: CaptureTarget | None | Any = CaptureTarget.STRING, stderr: CaptureTarget | None | Any = CaptureTarget.STRING) → collections.abc.Generator[Any, None, None]#

Capture outputs of sys.stdout and sys.stderr.

If stdout is STRING, capture sys.stdout as a string, if stdout is None, do not capture sys.stdout, leaving it untouched, otherwise redirect sys.stdout to the file-like object given by stdout.

Behave correspondingly for stderr with the exception that if stderr is STDOUT, redirect sys.stderr to stdout target and set stderr attribute of yielded object to None.

>>> from conda.common.io import captured
>>> with captured() as c:
...     print("hello world!")
...
>>> c.stdout
'hello world!\n'

Parameters:

stdout -- capture target for sys.stdout, one of STRING, None, or file-like object
stderr -- capture target for sys.stderr, one of STRING, STDOUT, None, or file-like object

Yields:

CapturedText --

has attributes stdout, stderr which are either strings, None or the: corresponding file-like function argument.

argv(args_list: list[str]) → collections.abc.Generator[None, None, None]#

Temporarily replace sys.argv with the given list.

Parameters:: args_list -- List of arguments to use as sys.argv.

disable_logger(logger_name: str) → collections.abc.Generator[None, None, None]#

Temporarily disable a logger by setting its level above CRITICAL.

Parameters:: logger_name -- Name of the logger to disable.

stderr_log_level(level: int, logger_name: str | None = None) → collections.abc.Generator[None, None, None]#

Temporarily set a logger to output to stderr at the given level.

Parameters:

level -- Logging level (e.g., logging.INFO).
logger_name -- Name of the logger; if None, uses the root logger.

attach_stderr_handler(level: int = WARN, logger_name: str | None = None, propagate: bool = False, formatter: logging.Formatter | None = None, filters: collections.abc.Iterable[Any] | None = None) → None#

Attach a new stderr handler to the given logger and configure both.

This function creates a new StreamHandler that writes to stderr and attaches it to the logger given by logger_name (which maybe None, in which case the root logger is used). If the logger already has a handler by the name of stderr, it is removed first.

The given level is set for the handler, not for the logger; however, this function also sets the level of the given logger to the minimum of its current effective level and the new handler level, ensuring that the handler will receive the required log records, while minimizing the number of unnecessary log events. It also sets the loggers propagate property according to the propagate argument. The formatter argument can be used to set the formatter of the handler.

timeout(timeout_secs: int | float, func: collections.abc.Callable[Ellipsis, Any], *args: Any, default_return: Any = None, **kwargs: Any) → Any#

Enforce a maximum time for a callable to complete.

Not yet implemented on Windows.

Parameters:

timeout_secs -- Maximum seconds to allow the callable to run.
func -- Callable to execute.
args -- Positional arguments to pass to the callable.
default_return -- Value to return if timeout or KeyboardInterrupt occurs.
kwargs -- Keyword arguments to pass to the callable.

Returns:

Return value of the callable, or default_return on timeout/interrupt.

class DummyExecutor#

Bases: concurrent.futures.Executor

Synchronous executor for debugging; executes tasks in the current thread.

_shutdown = False#

_shutdownLock#

submit(fn: collections.abc.Callable[Ellipsis, Any], *args: Any, **kwargs: Any) → concurrent.futures.Future[Any]#

Submits a callable to be executed with the given arguments.

Schedules the callable to be executed as fn(*args, **kwargs) and returns a Future instance representing the execution of the callable.

Returns:: A Future representing the given call.

map(func: collections.abc.Callable[Ellipsis, Any], *iterables: collections.abc.Iterable[Any]) → collections.abc.Iterator[Any]#: Map function over iterables, yielding results one at a time.

shutdown(wait: bool = True) → None#

Clean-up the resources associated with the Executor.

It is safe to call this method several times. Otherwise, no other methods can be called after this one.

Parameters:

wait -- If True then shutdown will not return until all running futures have finished executing and the resources used by the executor have been reclaimed.
cancel_futures -- If True then shutdown will cancel all pending futures. Futures that are completed or running will not be cancelled.

class ThreadLimitedThreadPoolExecutor(max_workers: int = 10)#

Bases: concurrent.futures.ThreadPoolExecutor

Thread pool executor that gracefully handles thread creation limits.

_adjust_thread_count() → None#

as_completed#

get_instrumentation_record_file() → str#

Return the path to the instrumentation record file.

Returns:: Expanded path from CONDA_INSTRUMENTATION_RECORD_FILE or default location.

class time_recorder(entry_name: str | None = None, module_name: str | None = None)#

Bases: ContextDecorator

Context decorator for recording execution times to a CSV file.

record_file = b'.'#

start_time: float | None = None#

total_call_num: dict[str, int]#

total_run_time: dict[str, float]#

entry_name = None#

module_name = None#

_set_entry_name(f: collections.abc.Callable[Ellipsis, Any]) → None#

__call__(f: collections.abc.Callable[Ellipsis, Any]) → collections.abc.Callable[Ellipsis, Any]#

__enter__() → time_recorder#

__exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) → None#

classmethod log_totals() → None#

_ensure_dir() → None#

print_instrumentation_data() → None#: Print aggregated instrumentation data from the record file as JSON.

io#

Classes#

Functions#

Attributes#

`io`#