FM / amqp

/opt/imh-python/lib/python3.9/site-packages/amqp/

/opt/imh-python/lib/python3.9/site-packages/amqp

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
abstract_channel.py	4941	0644	edit dl rm
basic_message.py	3357	0644	edit dl rm
channel.py	74475	0644	edit dl rm
connection.py	27541	0644	edit dl rm
exceptions.py	7166	0644	edit dl rm
method_framing.py	6734	0644	edit dl rm
platform.py	2379	0644	edit dl rm
protocol.py	291	0644	edit dl rm
sasl.py	5986	0644	edit dl rm
serialization.py	17162	0644	edit dl rm
spec.py	2121	0644	edit dl rm
transport.py	23046	0644	edit dl rm
utils.py	1456	0644	edit dl rm
__init__.py	2357	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/amqp/utils.py (1456B)

"""Compatibility utilities."""
import logging
from logging import NullHandler

# enables celery 3.1.23 to start again
from vine import promise  # noqa
from vine.utils import wraps

try:
    import fcntl
except ImportError:  # pragma: no cover
    fcntl = None  # noqa

def set_cloexec(fd, cloexec):
    """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)

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

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

def get_logger(logger):
    """Get logger by name."""
    if isinstance(logger, str):
        logger = logging.getLogger(logger)
    if not logger.handlers:
        logger.addHandler(NullHandler())
    return logger

"""Compatibility utilities.""" import logging from logging import NullHandler # enables celery 3.1.23 to start again from vine import promise # noqa from vine.utils import wraps try: import fcntl except ImportError: # pragma: no cover fcntl = None # noqa def set_cloexec(fd, cloexec): """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) 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 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 def get_logger(logger): """Get logger by name.""" if isinstance(logger, str): logger = logging.getLogger(logger) if not logger.handlers: logger.addHandler(NullHandler()) return logger