Source code for taskflow.utils.kombu_utils

# -*- coding: utf-8 -*-

#    Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

# Keys extracted from the message properties when formatting...
_MSG_PROPERTIES = tuple([
    'correlation_id',
    'delivery_info/routing_key',
    'type',
])


[docs]class DelayedPretty(object):
    """Wraps a message and delays prettifying it until requested.

    TODO(harlowja): remove this when https://github.com/celery/kombu/pull/454/
    is merged and a release is made that contains it (since that pull
    request is equivalent and/or better than this).
    """

    def __init__(self, message):
        self._message = message
        self._message_pretty = None

    def __str__(self):
        if self._message_pretty is None:
            self._message_pretty = _prettify_message(self._message)
        return self._message_pretty


def _get_deep(properties, *keys):
    """Get a final key among a list of keys (each with its own sub-dict)."""
    for key in keys:
        properties = properties[key]
    return properties


def _prettify_message(message):
    """Kombu doesn't currently have a useful ``__str__()`` or ``__repr__()``.

    This provides something decent(ish) for debugging (or other purposes) so
    that messages are more nice and understandable....
    """
    if message.content_type is not None:
        properties = {
            'content_type': message.content_type,
        }
    else:
        properties = {}
    for name in _MSG_PROPERTIES:
        segments = name.split("/")
        try:
            value = _get_deep(message.properties, *segments)
        except (KeyError, ValueError, TypeError):
            pass
        else:
            if value is not None:
                properties[segments[-1]] = value
    if message.body is not None:
        properties['body_length'] = len(message.body)
    return "%(delivery_tag)s: %(properties)s" % {
        'delivery_tag': message.delivery_tag,
        'properties': properties,
    }