This document is for py-amqp's development version, which can be significantly different from previous releases. Get the stable docs here: 2.4.
Source code for amqp.utils
"""Compatibility utilities."""
from __future__ import absolute_import, unicode_literals
import logging
# enables celery 3.1.23 to start again
from vine import promise # noqa
from vine.utils import wraps
from .five import PY3, string_t, text_t
try:
import fcntl
except ImportError: # pragma: no cover
fcntl = None # noqa
try:
from os import set_cloexec # Python 3.4?
except ImportError: # pragma: no cover
# TODO: Drop this once we drop Python 2.7 support
[docs] def set_cloexec(fd, cloexec): # noqa
"""Set flag to close fd after exec."""
if fcntl is None:
return
try:
FD_CLOEXEC = fcntl.FD_CLOEXEC
except AttributeError:
raise NotImplementedError(
'close-on-exec flag not supported on this platform',
)
flags = fcntl.fcntl(fd, fcntl.F_GETFD)
if cloexec:
flags |= FD_CLOEXEC
else:
flags &= ~FD_CLOEXEC
return fcntl.fcntl(fd, fcntl.F_SETFD, flags)
[docs]def get_errno(exc):
"""Get exception errno (if set).
Notes:
:exc:`socket.error` and :exc:`IOError` first got
the ``.errno`` attribute in Py2.7.
"""
try:
return exc.errno
except AttributeError:
try:
# e.args = (errno, reason)
if isinstance(exc.args, tuple) and len(exc.args) == 2:
return exc.args[0]
except AttributeError:
pass
return 0
[docs]def coro(gen):
"""Decorator to mark generator as a co-routine."""
@wraps(gen)
def _boot(*args, **kwargs):
co = gen(*args, **kwargs)
next(co)
return co
return _boot
if PY3: # pragma: no cover
def str_to_bytes(s):
"""Convert str to bytes."""
if isinstance(s, str):
return s.encode('utf-8', 'surrogatepass')
return s
def bytes_to_str(s):
"""Convert bytes to str."""
if isinstance(s, bytes):
return s.decode('utf-8', 'surrogatepass')
return s
else:
# TODO: Drop this once we drop Python 2.7 support
[docs] def str_to_bytes(s): # noqa
"""Convert str to bytes."""
if isinstance(s, text_t):
return s.encode('utf-8')
return s
[docs]def get_logger(logger):
"""Get logger by name."""
if isinstance(logger, string_t):
logger = logging.getLogger(logger)
if not logger.handlers:
logger.addHandler(NullHandler())
return logger