:py:mod:`io` ============ .. py:module:: conda.common.io .. autoapi-nested-parse:: Common I/O utilities. Classes ------- .. autoapisummary:: conda.common.io.DeltaSecondsFormatter conda.common.io.ContextDecorator conda.common.io.SwallowBrokenPipe conda.common.io.CaptureTarget conda.common.io.time_recorder Functions --------- .. autoapisummary:: conda.common.io.dashlist conda.common.io.env_vars conda.common.io.env_var conda.common.io.env_unmodified conda.common.io.captured conda.common.io.argv conda.common.io.disable_logger conda.common.io.stderr_log_level conda.common.io.attach_stderr_handler conda.common.io.timeout conda.common.io._load_concurrency conda.common.io.__getattr__ conda.common.io.get_instrumentation_record_file conda.common.io.print_instrumentation_data Attributes ---------- .. autoapisummary:: conda.common.io._FORMATTER conda.common.io.swallow_broken_pipe conda.common.io._LAZY_CONCURRENCY .. py:class:: DeltaSecondsFormatter(fmt: str | None = None, datefmt: str | None = None) Bases: :py:obj:`logging.Formatter` Logging formatter with additional attributes for run time logging. .. attribute:: `delta_secs` Elapsed seconds since last log/format call (or creation of logger). .. attribute:: `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. :param fmt: Log message format string. :param datefmt: Date format string. .. py:attribute:: prev_time .. py:method:: 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. .. py:data:: _FORMATTER .. py:function:: dashlist(iterable: collections.abc.Iterable[Any], indent: int = 2) -> str Format an iterable as a dashed list with optional indentation. :param iterable: Items to format. :param indent: Number of spaces for indentation. :returns: Formatted string with each item on a new line prefixed by ``- ``. .. py:class:: ContextDecorator Base class for a context manager class (implementing __enter__() and __exit__()) that also makes it a decorator. .. py:method:: __call__(f: collections.abc.Callable[..., Any]) -> collections.abc.Callable[..., Any] .. py:class:: SwallowBrokenPipe Bases: :py:obj:`ContextDecorator` Ignore BrokenPipeError and errors related to stdout or stderr being closed by a downstream program. .. py:method:: __enter__() -> None .. py:method:: __exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> bool | None .. py:data:: swallow_broken_pipe .. py:class:: CaptureTarget Bases: :py:obj:`enum.Enum` Constants used for contextmanager captured. Used similarly like the constants PIPE, STDOUT for stdlib's subprocess.Popen. .. py:attribute:: STRING :value: -1 .. py:attribute:: STDOUT :value: -2 .. py:function:: 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. :param var_map: Dictionary of environment variable names to values. :param callback: Optional callback invoked when entering and exiting the context. :param stack_callback: Optional callback invoked with True when entering, False when exiting. .. py:function:: 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. :param name: Environment variable name. :param value: Environment variable value. :param callback: Optional callback invoked when entering and exiting the context. :param stack_callback: Optional callback invoked with True when entering, False when exiting. .. py:function:: env_unmodified(callback: collections.abc.Callable[[], None] | None = None) -> collections.abc.Generator[None, None, None] Context manager that ensures environment variables are not modified. :param callback: Optional callback invoked when entering and exiting the context. .. py:function:: 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. .. code-block:: pycon >>> from conda.common.io import captured >>> with captured() as c: ... print("hello world!") ... >>> c.stdout 'hello world!\n' :param stdout: capture target for sys.stdout, one of STRING, None, or file-like object :param 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. .. py:function:: argv(args_list: list[str]) -> collections.abc.Generator[None, None, None] Temporarily replace sys.argv with the given list. :param args_list: List of arguments to use as sys.argv. .. py:function:: disable_logger(logger_name: str) -> collections.abc.Generator[None, None, None] Temporarily disable a logger by setting its level above CRITICAL. :param logger_name: Name of the logger to disable. .. py:function:: 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. :param level: Logging level (e.g., :const:`logging.INFO`). :param logger_name: Name of the logger; if None, uses the root logger. .. py:function:: attach_stderr_handler(level: int = WARNING, 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. .. py:function:: timeout(timeout_secs: int | float, func: collections.abc.Callable[..., Any], *args: Any, default_return: Any = None, **kwargs: Any) -> Any Enforce a maximum time for a callable to complete. Not yet implemented on Windows. :param timeout_secs: Maximum seconds to allow the callable to run. :param func: Callable to execute. :param args: Positional arguments to pass to the callable. :param default_return: Value to return if timeout or KeyboardInterrupt occurs. :param kwargs: Keyword arguments to pass to the callable. :returns: Return value of the callable, or ``default_return`` on timeout/interrupt. .. py:function:: _load_concurrency() -> None Lazily define executor classes and as_completed on first access. Importing concurrent.futures + threading costs ~45 modules. Deferring this to first use keeps ``import conda.common.io`` lightweight for code paths that never use parallel I/O (dashlist, time_recorder, etc.). .. py:data:: _LAZY_CONCURRENCY .. py:function:: __getattr__(name: str) .. py:function:: get_instrumentation_record_file() -> str Return the path to the instrumentation record file. :returns: Expanded path from ``CONDA_INSTRUMENTATION_RECORD_FILE`` or default location. .. py:class:: time_recorder(entry_name: str | None = None, module_name: str | None = None) Bases: :py:obj:`ContextDecorator` Context decorator for recording execution times to a CSV file. .. py:attribute:: record_file :value: b'.' .. py:attribute:: start_time :type: float | None :value: None .. py:attribute:: total_call_num :type: dict[str, int] .. py:attribute:: total_run_time :type: dict[str, float] .. py:attribute:: entry_name :value: None .. py:attribute:: module_name :value: None .. py:method:: _set_entry_name(f: collections.abc.Callable[..., Any]) -> None .. py:method:: __call__(f: collections.abc.Callable[..., Any]) -> collections.abc.Callable[..., Any] .. py:method:: __enter__() -> time_recorder .. py:method:: __exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None .. py:method:: log_totals() -> None :classmethod: .. py:method:: _ensure_dir() -> None .. py:function:: print_instrumentation_data() -> None Print aggregated instrumentation data from the record file as JSON.