Flow¶
-
@
silent
¶ Ignore all real exceptions (descendants of
Exception
). Handy for cleaning data such as user input:brand_id = silent(int)(request.GET['brand_id']) ids = keep(silent(int), request.GET.getlist('id'))
And in data import/transform:
get_greeting = compose(silent(string.lower), re_finder(r'(\w+)!')) map(get_greeting, ['a!', ' B!', 'c.']) # -> ['a', 'b', None]
Note
Avoid silencing non-primitive functions, use
@ignore()
instead and even then be careful not to swallow exceptions unintentionally.
-
@
ignore
(errors, default=None)¶ Same as
@silent
, but able to specifyerrors
to catch anddefault
to return in case of error caught.errors
can either be exception class or tuple of them.
-
suppress
(*errors)¶ A context manager which suppresses given exceptions under its scope:
with suppress(HttpError): # Assume this request can fail, and we are ok with it make_http_request()
-
@
once
¶ -
@
once_per_args
¶ -
@
once_per
(*argnames)¶ Call function only once, once for every combination of values of its arguments or once for every combination of given arguments. Thread safe. Handy for various initialization purposes:
# Global initialization @once def initialize_cache(): conn = some.Connection(...) # ... set up everything # Per argument initialization @once_per_args def initialize_language(lang): conf = load_language_conf(lang) # ... set up language # Setup each class once class SomeManager(Manager): @once_per('cls') def _initialize_class(self, cls): pre_save.connect(self._pre_save, sender=cls) # ... set up signals, no dups
-
raiser
(exception_or_class=Exception, *args, **kwargs)¶ Constructs function that raises given exception with given arguments on any invocation.
-
@
retry
(tries, errors=Exception, timeout=0)¶ Every call of the decorated function is tried up to
tries
times. The first attempt counts as a try. Retries occur when any subclass oferrors
is raised (errors
can be an exception class or a list/tuple of exception classes). There will be a delay intimeout
seconds between tries.A common use is to wrap some unreliable action:
@retry(3, errors=HttpError) def download_image(url): # ... make http request return image
You can pass callable as
timeout
to achieve exponential delays or other complex behavior:@retry(3, errors=HttpError, timeout=lambda a: 2 ** a) def download_image(url): # ... make http request return image
-
fallback
(*approaches)¶ Tries several approaches until one works. Each approach is either callable or a tuple
(callable, errors)
, where errors is an exception class or a tuple of classes, which signal to fall back to next approach. Iferrors
is not supplied then fall back is done for anyException
:fallback( (partial(send_mail, ADMIN_EMAIL, message), SMTPException), partial(log.error, message), raiser(FeedbackError, "Unable to notify admin") )
-
limit_error_rate
(fails, timeout, exception=ErrorRateExceeded)¶ If function fails to complete
fails
times in a row, calls to it will be intercepted fortimeout
withexception
raised instead. A clean way to short-circuit function taking too long to fail:@limit_error_rate(fails=5, timeout=60, exception=RequestError('Temporary unavailable')) def do_request(query): # ... make a http request return data
Can be combined with
ignore()
to silently stop trying for a while:@ignore(ErrorRateExceeded, default={'id': None, 'name': 'Unknown'}) @limit_error_rate(fails=5, timeout=60) def get_user(id): # ... make a http request return data
-
@
collecting
¶ Transforms generator or other iterator returning function into list returning one.
Handy to prevent quirky iterator-returning properties:
@property @collecting def path_up(self): node = self while node: yield node node = node.parent
Also makes list constructing functions beautifully yielding.
-
@
joining
(sep)¶ Wraps common python idiom “collect then join” into a decorator. Transforms generator or alike into function, returning string of joined results. Automatically converts all elements to separator type for convenience.
Goes well with generators with some ad-hoc logic within:
@joining(', ') def car_desc(self): yield self.year_made if self.engine_volume: yield '%s cc' % self.engine_volume if self.transmission: yield self.get_transmission_display() if self.gear: yield self.get_gear_display() # ...
Use
unicode
separator to get unicode result:@joining(u', ') def car_desc(self): yield self.year_made # ...
See also
str_join()
.
-
@
post_processing
(func)¶ Passes decorated function result through
func
. This is the generalization of@collecting
and@joining()
. Could save you writing a decorator or serve as extended comprehensions:@post_processing(dict) def make_cond(request): if request.GET['new']: yield 'year__gt', 2000 for key, value in request.GET.items(): if value == '': continue # ...