This document describes the current stable version of Celery (3.1). For development docs, go here.

Source code for kombu.utils.compat

"""
kombu.utils.compat
==================

Helps compatibility with older Python versions.

"""
from __future__ import absolute_import


# ############# timedelta_seconds() -> delta.total_seconds ###################
from datetime import timedelta

HAVE_TIMEDELTA_TOTAL_SECONDS = hasattr(timedelta, 'total_seconds')


if HAVE_TIMEDELTA_TOTAL_SECONDS:   # pragma: no cover

    def timedelta_seconds(delta):
        """Convert :class:`datetime.timedelta` to seconds.

        Doesn't account for negative values.

        """
        return max(delta.total_seconds(), 0)

else:  # pragma: no cover

    def timedelta_seconds(delta):  # noqa
        """Convert :class:`datetime.timedelta` to seconds.

        Doesn't account for negative values.

        """
        if delta.days < 0:
            return 0
        return delta.days * 86400 + delta.seconds + (delta.microseconds / 10e5)

# ############# socket.error.errno ###########################################


def get_errno(exc):
    """: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

# ############# collections.OrderedDict ######################################
try:
    from collections import OrderedDict
except ImportError:
    from ordereddict import OrderedDict  # noqa