Module materialize.ui
Utilities for interacting with humans.
Expand source code Browse git
# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.
"""Utilities for interacting with humans."""
import asyncio
import datetime
import os
import shlex
import sys
import time
from collections.abc import AsyncGenerator, Callable, Generator, Iterable
from contextlib import contextmanager
from typing import Any
from colored import attr, fg
class Verbosity:
"""How noisy logs should be"""
quiet: bool = False
@classmethod
def init_from_env(cls, explicit: bool | None) -> None:
"""Set to quiet based on MZ_QUIET being set to almost any value
The only values that this gets set to false for are the empty string, 0, or no
"""
cls.quiet = env_is_truthy("MZ_QUIET")
if explicit is not None:
cls.quiet = explicit
def speaker(prefix: str) -> Callable[..., None]:
"""Create a function that will log with a prefix to stderr.
Obeys `Verbosity.quiet`. Note that you must include any necessary
spacing after the prefix.
Example::
>>> say = speaker("mz> ")
>>> say("hello") # doctest: +SKIP
mz> hello
"""
def say(msg: str) -> None:
if not Verbosity.quiet:
print(f"{prefix}{msg}", file=sys.stderr)
return say
header = speaker("==> ")
say = speaker("")
def warn(message: str) -> None:
"""Emits a warning message to stderr."""
print(f"{fg('yellow')}warning:{attr('reset')} {message}")
def confirm(question: str) -> bool:
"""Render a question, returning True if the user says y or yes"""
response = input(f"{question} [y/N]")
return response.lower() in ("y", "yes")
def progress(msg: str = "", prefix: str | None = None, *, finish: bool = False) -> None:
"""Print a progress message to stderr, using the same prefix format as speaker"""
if prefix is not None:
msg = f"{prefix}> {msg}"
end = "" if not finish else "\n"
print(msg, file=sys.stderr, flush=True, end=end)
def timeout_loop(timeout: int, tick: float = 1.0) -> Generator[float, None, None]:
"""Loop until timeout, optionally sleeping until tick
Always iterates at least once
Args:
timeout: maximum. number of seconds to wait
tick: how long to ensure passes between loop iterations. Default: 1
"""
end = time.monotonic() + timeout
while True:
before = time.monotonic()
yield end - before
after = time.monotonic()
if after >= end:
return
if after - before < tick:
if tick > 0:
time.sleep(tick - (after - before))
async def async_timeout_loop(
timeout: int, tick: float = 1.0
) -> AsyncGenerator[float, None]:
"""Loop until timeout, asynchronously sleeping until tick
Always iterates at least once
Args:
timeout: maximum. number of seconds to wait
tick: how long to ensure passes between loop iterations. Default: 1
"""
end = time.monotonic() + timeout
while True:
before = time.monotonic()
yield end - before
after = time.monotonic()
if after >= end:
return
if after - before < tick:
if tick > 0:
await asyncio.sleep(tick - (after - before))
def log_in_automation(msg: str) -> None:
"""Log to a file, if we're running in automation"""
if env_is_truthy("MZ_IN_AUTOMATION"):
with open("/tmp/mzcompose.log", "a") as fh:
now = datetime.datetime.now().isoformat()
print(f"[{now}] {msg}", file=fh)
def shell_quote(args: Iterable[Any]) -> str:
"""Return shell-escaped string of all the parameters
::
>>> shell_quote(["one", "two three"])
"one 'two three'"
"""
return " ".join(shlex.quote(str(arg)) for arg in args)
def env_is_truthy(env_var: str) -> bool:
"""Return true if `env_var` is set and is not one of: 0, '', no, false"""
env = os.getenv(env_var)
if env is not None:
return env not in ("", "0", "no", "false")
return False
class UIError(Exception):
"""An error intended for display to humans.
Use this exception type when the error is something the user can be expected
to handle. If the error indicates a truly unexpected condition (i.e., a
programming error), use a different exception type that will produce a
backtrace instead.
Attributes:
hint: An optional hint to display alongside the error message.
"""
def __init__(self, message: str, hint: str | None = None):
super().__init__(message)
self.hint = hint
def set_hint(self, hint: str) -> None:
"""Attaches a hint to the error.
This method will overwrite the existing hint, if any.
"""
self.hint = hint
class CommandFailureCausedUIError(UIError):
"""
An UIError that is caused by executing a command.
"""
def __init__(
self,
message: str,
cmd: list[str],
stdout: str | None = None,
stderr: str | None = None,
hint: str | None = None,
):
super().__init__(message, hint)
self.cmd = cmd
self.stdout = stdout
self.stderr = stderr
@contextmanager
def error_handler(prog: str) -> Any:
"""Catches and pretty-prints any raised `UIError`s.
Args:
prog: The name of the program with which to prefix the error message.
"""
try:
yield
except UIError as e:
print(f"{prog}: {fg('red')}error:{attr('reset')} {e}", file=sys.stderr)
if e.hint:
print(f"{attr('bold')}hint:{attr('reset')} {e.hint}")
sys.exit(1)
except KeyboardInterrupt:
sys.exit(1)
Functions
async def async_timeout_loop(timeout: int, tick: float = 1.0) ‑> collections.abc.AsyncGenerator[float, None]
-
Loop until timeout, asynchronously sleeping until tick
Always iterates at least once
Args
timeout
- maximum. number of seconds to wait
tick
- how long to ensure passes between loop iterations. Default: 1
Expand source code Browse git
async def async_timeout_loop( timeout: int, tick: float = 1.0 ) -> AsyncGenerator[float, None]: """Loop until timeout, asynchronously sleeping until tick Always iterates at least once Args: timeout: maximum. number of seconds to wait tick: how long to ensure passes between loop iterations. Default: 1 """ end = time.monotonic() + timeout while True: before = time.monotonic() yield end - before after = time.monotonic() if after >= end: return if after - before < tick: if tick > 0: await asyncio.sleep(tick - (after - before))
def confirm(question: str) ‑> bool
-
Render a question, returning True if the user says y or yes
Expand source code Browse git
def confirm(question: str) -> bool: """Render a question, returning True if the user says y or yes""" response = input(f"{question} [y/N]") return response.lower() in ("y", "yes")
def env_is_truthy(env_var: str) ‑> bool
-
Return true if
env_var
is set and is not one of: 0, '', no, falseExpand source code Browse git
def env_is_truthy(env_var: str) -> bool: """Return true if `env_var` is set and is not one of: 0, '', no, false""" env = os.getenv(env_var) if env is not None: return env not in ("", "0", "no", "false") return False
def error_handler(prog: str) ‑> Any
-
Catches and pretty-prints any raised
UIError
s.Args
prog
- The name of the program with which to prefix the error message.
Expand source code Browse git
@contextmanager def error_handler(prog: str) -> Any: """Catches and pretty-prints any raised `UIError`s. Args: prog: The name of the program with which to prefix the error message. """ try: yield except UIError as e: print(f"{prog}: {fg('red')}error:{attr('reset')} {e}", file=sys.stderr) if e.hint: print(f"{attr('bold')}hint:{attr('reset')} {e.hint}") sys.exit(1) except KeyboardInterrupt: sys.exit(1)
def header(msg: str) ‑> None
-
Expand source code Browse git
def say(msg: str) -> None: if not Verbosity.quiet: print(f"{prefix}{msg}", file=sys.stderr)
def log_in_automation(msg: str) ‑> None
-
Log to a file, if we're running in automation
Expand source code Browse git
def log_in_automation(msg: str) -> None: """Log to a file, if we're running in automation""" if env_is_truthy("MZ_IN_AUTOMATION"): with open("/tmp/mzcompose.log", "a") as fh: now = datetime.datetime.now().isoformat() print(f"[{now}] {msg}", file=fh)
def progress(msg: str = '', prefix: str | None = None, *, finish: bool = False) ‑> None
-
Print a progress message to stderr, using the same prefix format as speaker
Expand source code Browse git
def progress(msg: str = "", prefix: str | None = None, *, finish: bool = False) -> None: """Print a progress message to stderr, using the same prefix format as speaker""" if prefix is not None: msg = f"{prefix}> {msg}" end = "" if not finish else "\n" print(msg, file=sys.stderr, flush=True, end=end)
def say(msg: str) ‑> None
-
Expand source code Browse git
def say(msg: str) -> None: if not Verbosity.quiet: print(f"{prefix}{msg}", file=sys.stderr)
def shell_quote(args: collections.abc.Iterable[typing.Any]) ‑> str
-
Return shell-escaped string of all the parameters
::
>>> shell_quote(["one", "two three"]) "one 'two three'"
Expand source code Browse git
def shell_quote(args: Iterable[Any]) -> str: """Return shell-escaped string of all the parameters :: >>> shell_quote(["one", "two three"]) "one 'two three'" """ return " ".join(shlex.quote(str(arg)) for arg in args)
def speaker(prefix: str) ‑> collections.abc.Callable[..., None]
-
Create a function that will log with a prefix to stderr.
Obeys
Verbosity.quiet
. Note that you must include any necessary spacing after the prefix.Example::
>>> say = speaker("mz> ") >>> say("hello") # doctest: +SKIP mz> hello
Expand source code Browse git
def speaker(prefix: str) -> Callable[..., None]: """Create a function that will log with a prefix to stderr. Obeys `Verbosity.quiet`. Note that you must include any necessary spacing after the prefix. Example:: >>> say = speaker("mz> ") >>> say("hello") # doctest: +SKIP mz> hello """ def say(msg: str) -> None: if not Verbosity.quiet: print(f"{prefix}{msg}", file=sys.stderr) return say
def timeout_loop(timeout: int, tick: float = 1.0) ‑> collections.abc.Generator[float, None, None]
-
Loop until timeout, optionally sleeping until tick
Always iterates at least once
Args
timeout
- maximum. number of seconds to wait
tick
- how long to ensure passes between loop iterations. Default: 1
Expand source code Browse git
def timeout_loop(timeout: int, tick: float = 1.0) -> Generator[float, None, None]: """Loop until timeout, optionally sleeping until tick Always iterates at least once Args: timeout: maximum. number of seconds to wait tick: how long to ensure passes between loop iterations. Default: 1 """ end = time.monotonic() + timeout while True: before = time.monotonic() yield end - before after = time.monotonic() if after >= end: return if after - before < tick: if tick > 0: time.sleep(tick - (after - before))
def warn(message: str) ‑> None
-
Emits a warning message to stderr.
Expand source code Browse git
def warn(message: str) -> None: """Emits a warning message to stderr.""" print(f"{fg('yellow')}warning:{attr('reset')} {message}")
Classes
class CommandFailureCausedUIError (message: str, cmd: list[str], stdout: str | None = None, stderr: str | None = None, hint: str | None = None)
-
An UIError that is caused by executing a command.
Expand source code Browse git
class CommandFailureCausedUIError(UIError): """ An UIError that is caused by executing a command. """ def __init__( self, message: str, cmd: list[str], stdout: str | None = None, stderr: str | None = None, hint: str | None = None, ): super().__init__(message, hint) self.cmd = cmd self.stdout = stdout self.stderr = stderr
Ancestors
- UIError
- builtins.Exception
- builtins.BaseException
Inherited members
class UIError (message: str, hint: str | None = None)
-
An error intended for display to humans.
Use this exception type when the error is something the user can be expected to handle. If the error indicates a truly unexpected condition (i.e., a programming error), use a different exception type that will produce a backtrace instead.
Attributes
hint
- An optional hint to display alongside the error message.
Expand source code Browse git
class UIError(Exception): """An error intended for display to humans. Use this exception type when the error is something the user can be expected to handle. If the error indicates a truly unexpected condition (i.e., a programming error), use a different exception type that will produce a backtrace instead. Attributes: hint: An optional hint to display alongside the error message. """ def __init__(self, message: str, hint: str | None = None): super().__init__(message) self.hint = hint def set_hint(self, hint: str) -> None: """Attaches a hint to the error. This method will overwrite the existing hint, if any. """ self.hint = hint
Ancestors
- builtins.Exception
- builtins.BaseException
Subclasses
Methods
def set_hint(self, hint: str) ‑> None
-
Attaches a hint to the error.
This method will overwrite the existing hint, if any.
Expand source code Browse git
def set_hint(self, hint: str) -> None: """Attaches a hint to the error. This method will overwrite the existing hint, if any. """ self.hint = hint
class Verbosity
-
How noisy logs should be
Expand source code Browse git
class Verbosity: """How noisy logs should be""" quiet: bool = False @classmethod def init_from_env(cls, explicit: bool | None) -> None: """Set to quiet based on MZ_QUIET being set to almost any value The only values that this gets set to false for are the empty string, 0, or no """ cls.quiet = env_is_truthy("MZ_QUIET") if explicit is not None: cls.quiet = explicit
Class variables
var quiet : bool
Static methods
def init_from_env(explicit: bool | None) ‑> None
-
Set to quiet based on MZ_QUIET being set to almost any value
The only values that this gets set to false for are the empty string, 0, or no
Expand source code Browse git
@classmethod def init_from_env(cls, explicit: bool | None) -> None: """Set to quiet based on MZ_QUIET being set to almost any value The only values that this gets set to false for are the empty string, 0, or no """ cls.quiet = env_is_truthy("MZ_QUIET") if explicit is not None: cls.quiet = explicit