__init__.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380
  1. from __future__ import generators
  2. import time
  3. import operator
  4. try:
  5. import ctypes
  6. except ImportError:
  7. ctypes = None
  8. import importlib
  9. from datetime import datetime
  10. from uuid import UUID, uuid4, _uuid_generate_random
  11. from inspect import getargspec
  12. from itertools import islice
  13. from carrot.utils import rpartition
  14. from dateutil.parser import parse as parse_iso8601
  15. from celery.utils.compat import all, any, defaultdict
  16. from celery.utils.timeutils import timedelta_seconds # was here before
  17. from celery.utils.functional import curry
  18. class promise(object):
  19. """A promise.
  20. Evaluated when called or if the :meth:`evaluate` method is called.
  21. The function is evaluated on every access, so the value is not
  22. memoized (see :class:`mpromise`).
  23. Overloaded operations that will evaluate the promise:
  24. :meth:`__str__`, :meth:`__repr__`, :meth:`__cmp__`.
  25. """
  26. def __init__(self, fun, *args, **kwargs):
  27. self._fun = fun
  28. self._args = args
  29. self._kwargs = kwargs
  30. def __call__(self):
  31. return self.evaluate()
  32. def evaluate(self):
  33. return self._fun(*self._args, **self._kwargs)
  34. def __str__(self):
  35. return str(self())
  36. def __repr__(self):
  37. return repr(self())
  38. def __cmp__(self, rhs):
  39. if isinstance(rhs, self.__class__):
  40. return -cmp(rhs, self())
  41. return cmp(self(), rhs)
  42. def __deepcopy__(self, memo):
  43. memo[id(self)] = self
  44. return self
  45. def __reduce__(self):
  46. return (self.__class__, (self._fun, ), {"_args": self._args,
  47. "_kwargs": self._kwargs})
  48. class mpromise(promise):
  49. """Memoized promise.
  50. The function is only evaluated once, every subsequent access
  51. will return the same value.
  52. .. attribute:: evaluated
  53. Set to to :const:`True` after the promise has been evaluated.
  54. """
  55. evaluated = False
  56. _value = None
  57. def evaluate(self):
  58. if not self.evaluated:
  59. self._value = super(mpromise, self).evaluate()
  60. self.evaluated = True
  61. return self._value
  62. def maybe_promise(value):
  63. """Evaluates if the value is a promise."""
  64. if isinstance(value, promise):
  65. return value.evaluate()
  66. return value
  67. def noop(*args, **kwargs):
  68. """No operation.
  69. Takes any arguments/keyword arguments and does nothing.
  70. """
  71. pass
  72. def maybe_iso8601(dt):
  73. """``Either datetime | str -> datetime or None -> None``"""
  74. if not dt:
  75. return
  76. if isinstance(dt, datetime):
  77. return dt
  78. return parse_iso8601(dt)
  79. def kwdict(kwargs):
  80. """Make sure keyword arguments are not in unicode.
  81. This should be fixed in newer Python versions,
  82. see: http://bugs.python.org/issue4978.
  83. """
  84. return dict((key.encode("utf-8"), value)
  85. for key, value in kwargs.items())
  86. def first(predicate, iterable):
  87. """Returns the first element in ``iterable`` that ``predicate`` returns a
  88. ``True`` value for."""
  89. for item in iterable:
  90. if predicate(item):
  91. return item
  92. def firstmethod(method):
  93. """Returns a functions that with a list of instances,
  94. finds the first instance that returns a value for the given method.
  95. The list can also contain promises (:class:`promise`.)
  96. """
  97. def _matcher(seq, *args, **kwargs):
  98. for cls in seq:
  99. try:
  100. answer = getattr(maybe_promise(cls), method)(*args, **kwargs)
  101. if answer is not None:
  102. return answer
  103. except AttributeError:
  104. pass
  105. return _matcher
  106. def chunks(it, n):
  107. """Split an iterator into chunks with ``n`` elements each.
  108. Examples
  109. # n == 2
  110. >>> x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 2)
  111. >>> list(x)
  112. [[0, 1], [2, 3], [4, 5], [6, 7], [8, 9], [10]]
  113. # n == 3
  114. >>> x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 3)
  115. >>> list(x)
  116. [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10]]
  117. """
  118. for first in it:
  119. yield [first] + list(islice(it, n - 1))
  120. def gen_unique_id():
  121. """Generate a unique id, having - hopefully - a very small chance of
  122. collission.
  123. For now this is provided by :func:`uuid.uuid4`.
  124. """
  125. # Workaround for http://bugs.python.org/issue4607
  126. if ctypes and _uuid_generate_random:
  127. buffer = ctypes.create_string_buffer(16)
  128. _uuid_generate_random(buffer)
  129. return str(UUID(bytes=buffer.raw))
  130. return str(uuid4())
  131. def padlist(container, size, default=None):
  132. """Pad list with default elements.
  133. Examples:
  134. >>> first, last, city = padlist(["George", "Costanza", "NYC"], 3)
  135. ("George", "Costanza", "NYC")
  136. >>> first, last, city = padlist(["George", "Costanza"], 3)
  137. ("George", "Costanza", None)
  138. >>> first, last, city, planet = padlist(["George", "Costanza",
  139. "NYC"], 4, default="Earth")
  140. ("George", "Costanza", "NYC", "Earth")
  141. """
  142. return list(container)[:size] + [default] * (size - len(container))
  143. def is_iterable(obj):
  144. try:
  145. iter(obj)
  146. except TypeError:
  147. return False
  148. return True
  149. def mitemgetter(*items):
  150. """Like :func:`operator.itemgetter` but returns ``None`` on missing items
  151. instead of raising :exc:`KeyError`."""
  152. return lambda container: map(container.get, items)
  153. def mattrgetter(*attrs):
  154. """Like :func:`operator.itemgetter` but returns ``None`` on missing
  155. attributes instead of raising :exc:`AttributeError`."""
  156. return lambda obj: dict((attr, getattr(obj, attr, None))
  157. for attr in attrs)
  158. def get_full_cls_name(cls):
  159. """With a class, get its full module and class name."""
  160. return ".".join([cls.__module__,
  161. cls.__name__])
  162. def repeatlast(it):
  163. """Iterate over all elements in the iterator, and when its exhausted
  164. yield the last value infinitely."""
  165. for item in it:
  166. yield item
  167. while 1: # pragma: no cover
  168. yield item
  169. def retry_over_time(fun, catch, args=[], kwargs={}, errback=noop,
  170. max_retries=None, interval_start=2, interval_step=2, interval_max=30):
  171. """Retry the function over and over until max retries is exceeded.
  172. For each retry we sleep a for a while before we try again, this interval
  173. is increased for every retry until the max seconds is reached.
  174. :param fun: The function to try
  175. :param catch: Exceptions to catch, can be either tuple or a single
  176. exception class.
  177. :keyword args: Positional arguments passed on to the function.
  178. :keyword kwargs: Keyword arguments passed on to the function.
  179. :keyword errback: Callback for when an exception in ``catch`` is raised.
  180. The callback must take two arguments: ``exc`` and ``interval``, where
  181. ``exc`` is the exception instance, and ``interval`` is the time in
  182. seconds to sleep next..
  183. :keyword max_retries: Maximum number of retries before we give up.
  184. If this is not set, we will retry forever.
  185. :keyword interval_start: How long (in seconds) we start sleeping between
  186. retries.
  187. :keyword interval_step: By how much the interval is increased for each
  188. retry.
  189. :keyword interval_max: Maximum number of seconds to sleep between retries.
  190. """
  191. retries = 0
  192. interval_range = xrange(interval_start,
  193. interval_max + interval_start,
  194. interval_step)
  195. for interval in repeatlast(interval_range):
  196. try:
  197. retval = fun(*args, **kwargs)
  198. except catch, exc:
  199. if max_retries and retries > max_retries:
  200. raise
  201. errback(exc, interval)
  202. retries += 1
  203. time.sleep(interval)
  204. else:
  205. return retval
  206. def fun_takes_kwargs(fun, kwlist=[]):
  207. """With a function, and a list of keyword arguments, returns arguments
  208. in the list which the function takes.
  209. If the object has an ``argspec`` attribute that is used instead
  210. of using the :meth:`inspect.getargspec`` introspection.
  211. :param fun: The function to inspect arguments of.
  212. :param kwlist: The list of keyword arguments.
  213. Examples
  214. >>> def foo(self, x, y, logfile=None, loglevel=None):
  215. ... return x * y
  216. >>> fun_takes_kwargs(foo, ["logfile", "loglevel", "task_id"])
  217. ["logfile", "loglevel"]
  218. >>> def foo(self, x, y, **kwargs):
  219. >>> fun_takes_kwargs(foo, ["logfile", "loglevel", "task_id"])
  220. ["logfile", "loglevel", "task_id"]
  221. """
  222. argspec = getattr(fun, "argspec", getargspec(fun))
  223. args, _varargs, keywords, _defaults = argspec
  224. if keywords != None:
  225. return kwlist
  226. return filter(curry(operator.contains, args), kwlist)
  227. def get_cls_by_name(name, aliases={}):
  228. """Get class by name.
  229. The name should be the full dot-separated path to the class::
  230. modulename.ClassName
  231. Example::
  232. celery.concurrency.processes.TaskPool
  233. ^- class name
  234. If ``aliases`` is provided, a dict containing short name/long name
  235. mappings, the name is looked up in the aliases first.
  236. Examples:
  237. >>> get_cls_by_name("celery.concurrency.processes.TaskPool")
  238. <class 'celery.concurrency.processes.TaskPool'>
  239. >>> get_cls_by_name("default", {
  240. ... "default": "celery.concurrency.processes.TaskPool"})
  241. <class 'celery.concurrency.processes.TaskPool'>
  242. # Does not try to look up non-string names.
  243. >>> from celery.concurrency.processes import TaskPool
  244. >>> get_cls_by_name(TaskPool) is TaskPool
  245. True
  246. """
  247. if not isinstance(name, basestring):
  248. return name # already a class
  249. name = aliases.get(name) or name
  250. module_name, _, cls_name = rpartition(name, ".")
  251. module = importlib.import_module(module_name)
  252. return getattr(module, cls_name)
  253. def instantiate(name, *args, **kwargs):
  254. """Instantiate class by name.
  255. See :func:`get_cls_by_name`.
  256. """
  257. return get_cls_by_name(name)(*args, **kwargs)
  258. def truncate_text(text, maxlen=128, suffix="..."):
  259. """Truncates text to a maximum number of characters."""
  260. if len(text) >= maxlen:
  261. return text[:maxlen].rsplit(" ", 1)[0] + suffix
  262. return text
  263. def abbr(S, max, ellipsis="..."):
  264. if S is None:
  265. return "???"
  266. if len(S) > max:
  267. return ellipsis and (S[:max-len(ellipsis)] + ellipsis) or S[:max]
  268. return S
  269. def abbrtask(S, max):
  270. if S is None:
  271. return "???"
  272. if len(S) > max:
  273. module, _, cls = rpartition(S, ".")
  274. module = abbr(module, max - len(cls), False)
  275. return module + "[.]" + cls
  276. return S