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
[docs] 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