codekingpro/portable-devtools

1from __future__ import annotations2 3from collections.abc import Iterable4from contextlib import contextmanager5import logging6import sys7import textwrap8from typing import Iterator9from typing import Optional10from typing import TextIO11from typing import Union12import warnings13 14from sqlalchemy.engine import url15 16from . import sqla_compat17 18log = logging.getLogger(__name__)19 20# disable "no handler found" errors21logging.getLogger("alembic").addHandler(logging.NullHandler())22 23 24try:25    import fcntl26    import termios27    import struct28 29    ioctl = fcntl.ioctl(0, termios.TIOCGWINSZ, struct.pack("HHHH", 0, 0, 0, 0))30    _h, TERMWIDTH, _hp, _wp = struct.unpack("HHHH", ioctl)31    if TERMWIDTH <= 0:  # can occur if running in emacs pseudo-tty32        TERMWIDTH = None33except (ImportError, OSError):34    TERMWIDTH = None35 36 37def write_outstream(38    stream: TextIO, *text: Union[str, bytes], quiet: bool = False39) -> None:40    if quiet:41        return42    encoding = getattr(stream, "encoding", "ascii") or "ascii"43    for t in text:44        if not isinstance(t, bytes):45            t = t.encode(encoding, "replace")46        t = t.decode(encoding)47        try:48            stream.write(t)49        except OSError:50            # suppress "broken pipe" errors.51            # no known way to handle this on Python 3 however52            # as the exception is "ignored" (noisily) in TextIOWrapper.53            break54 55 56@contextmanager57def status(58    status_msg: str, newline: bool = False, quiet: bool = False59) -> Iterator[None]:60    msg(status_msg + " ...", newline, flush=True, quiet=quiet)61    try:62        yield63    except:64        if not quiet:65            write_outstream(sys.stdout, "  FAILED\n")66        raise67    else:68        if not quiet:69            write_outstream(sys.stdout, "  done\n")70 71 72def err(message: str, quiet: bool = False) -> None:73    log.error(message)74    msg(f"FAILED: {message}", quiet=quiet)75    sys.exit(-1)76 77 78def obfuscate_url_pw(input_url: str) -> str:79    u = url.make_url(input_url)80    return sqla_compat.url_render_as_string(u, hide_password=True)  # type: ignore  # noqa: E50181 82 83def warn(msg: str, stacklevel: int = 2) -> None:84    warnings.warn(msg, UserWarning, stacklevel=stacklevel)85 86 87def msg(88    msg: str, newline: bool = True, flush: bool = False, quiet: bool = False89) -> None:90    if quiet:91        return92    if TERMWIDTH is None:93        write_outstream(sys.stdout, msg)94        if newline:95            write_outstream(sys.stdout, "\n")96    else:97        # left indent output lines98        lines = textwrap.wrap(msg, TERMWIDTH)99        if len(lines) > 1:100            for line in lines[0:-1]:101                write_outstream(sys.stdout, "  ", line, "\n")102        write_outstream(sys.stdout, "  ", lines[-1], ("\n" if newline else ""))103    if flush:104        sys.stdout.flush()105 106 107def format_as_comma(value: Optional[Union[str, Iterable[str]]]) -> str:108    if value is None:109        return ""110    elif isinstance(value, str):111        return value112    elif isinstance(value, Iterable):113        return ", ".join(value)114    else:115        raise ValueError("Don't know how to comma-format %r" % value)116