Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/alembic/util/messaging.py : 36%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import logging
2import sys
3import textwrap
4import warnings
6from sqlalchemy.engine import url
8from .compat import binary_type
9from .compat import collections_abc
10from .compat import py27
11from .compat import string_types
13log = logging.getLogger(__name__)
15if py27:
16 # disable "no handler found" errors
17 logging.getLogger("alembic").addHandler(logging.NullHandler())
20try:
21 import fcntl
22 import termios
23 import struct
25 ioctl = fcntl.ioctl(0, termios.TIOCGWINSZ, struct.pack("HHHH", 0, 0, 0, 0))
26 _h, TERMWIDTH, _hp, _wp = struct.unpack("HHHH", ioctl)
27 if TERMWIDTH <= 0: # can occur if running in emacs pseudo-tty
28 TERMWIDTH = None
29except (ImportError, IOError):
30 TERMWIDTH = None
33def write_outstream(stream, *text):
34 encoding = getattr(stream, "encoding", "ascii") or "ascii"
35 for t in text:
36 if not isinstance(t, binary_type):
37 t = t.encode(encoding, "replace")
38 t = t.decode(encoding)
39 try:
40 stream.write(t)
41 except IOError:
42 # suppress "broken pipe" errors.
43 # no known way to handle this on Python 3 however
44 # as the exception is "ignored" (noisily) in TextIOWrapper.
45 break
48def status(_statmsg, fn, *arg, **kw):
49 newline = kw.pop("newline", False)
50 msg(_statmsg + " ...", newline, True)
51 try:
52 ret = fn(*arg, **kw)
53 write_outstream(sys.stdout, " done\n")
54 return ret
55 except:
56 write_outstream(sys.stdout, " FAILED\n")
57 raise
60def err(message):
61 log.error(message)
62 msg("FAILED: %s" % message)
63 sys.exit(-1)
66def obfuscate_url_pw(u):
67 u = url.make_url(u)
68 if u.password:
69 u.password = "XXXXX"
70 return str(u)
73def warn(msg, stacklevel=2):
74 warnings.warn(msg, UserWarning, stacklevel=stacklevel)
77def msg(msg, newline=True, flush=False):
78 if TERMWIDTH is None:
79 write_outstream(sys.stdout, msg)
80 if newline:
81 write_outstream(sys.stdout, "\n")
82 else:
83 # left indent output lines
84 lines = textwrap.wrap(msg, TERMWIDTH)
85 if len(lines) > 1:
86 for line in lines[0:-1]:
87 write_outstream(sys.stdout, " ", line, "\n")
88 write_outstream(sys.stdout, " ", lines[-1], ("\n" if newline else ""))
89 if flush:
90 sys.stdout.flush()
93def format_as_comma(value):
94 if value is None:
95 return ""
96 elif isinstance(value, string_types):
97 return value
98 elif isinstance(value, collections_abc.Iterable):
99 return ", ".join(value)
100 else:
101 raise ValueError("Don't know how to comma-format %r" % value)