Commodity’s docs¶
commodity
is a library of recurrent required utilities for Python programmers.
Contents:
deco¶
-
commodity.deco.
add_attr
(name, value)[source]¶ Set the given function attribute:
>>> @add_attr('timeout', 4) ... def func(args): ... pass ... >>> func.timeout 4
-
commodity.deco.
handle_exception
(exception, handler)[source]¶ Add and exception handler with decorator
>>> import logging, math >>> @handle_exception(ValueError, logging.error) ... def silent_sqrt(value): ... return math.sqrt(value) >>> silent_sqrt(-1) ERROR:root:math domain error
-
commodity.deco.
suppress_errors
(func=None, default=None, log_func=<function debug>, errors=[<type 'exceptions.Exception'>])[source]¶ Decorator that avoids to create try/excepts only to avoid valid errors. It can return a default value and ignore only some exceptions.
Example: It will return the file content or the empty string if it doesn’t exists:
>>> @suppress_errors(errors=[IOError], default='') ... def read_file(filename): ... with file(filename) as f: ... return f.read()
>>> read_file('/some/not/exists') ''
Parameters: - func – Function to be called.
- default – Default value to be returned if any of the exceptions are launched.
- log_func – The log function. logging.debug by default. It is a good idea to use it to avoid ‘catch pokemon’ problems.
- errors – list of allowed Exceptions. It takes in mind inheritance.
args¶
Creating a powerful argument parser is very easy:
from commodity.args import parser, add_argument, args
add_argument('-f', '--foo', help='foo argument')
add_argument('-v', '--verbose', action='store_true', help='be verbose')
parser.parse_args('--foo 3'.split())
assert args.foo == 3
And it supports config files. That requires argument specs (config.specs
):
[ui]
foo = integer(default=0)
verbose = boolean(default=False)
…the user config file (.app.config
):
[ui]
foo = 4
verbose = True
…and the code:
from commodity.args import parser, add_argument, args
add_argument('-f', '--foo', help='foo argument')
add_argument('-v', '--verbose', action='store_true', help='be verbose')
parser.set_specs("config.specs")
parser.load_config("/etc/app.config")
parser.load_config("/home/user/.app.config")
parser.load_config("by-dir.config")
parser.parse_args()
assert args.foo == 4
log¶
-
class
commodity.log.
CallerData
(depth=0)[source]¶ Get info about caller function, file, line and statement
>>> def a(): ... return b() >>> >>> def b(): ... print CallerData() >>> >>> a() File "<stdin>", line 2, in a() return b()
-
class
commodity.log.
CapitalLoggingFormatter
(fmt=None, datefmt=None)[source]¶ Define variable “levelcapital” for message formating. You can do things like: [EE] foo bar
>>> formatter = CapitalLoggingFormatter('[%(levelcapital)s] %(message)s') >>> console = logging.StreamHandler() >>> console.setFormatter(formatter)
-
class
commodity.log.
UniqueFilter
(name='')[source]¶ Log messages are allowed through just once. The ‘message’ includes substitutions, but is not formatted by the handler. If it were, then practically all messages would be unique!
from: http://code.activestate.com/recipes/412552-using-the-logging-module/
mail¶
path¶
-
commodity.path.
abs_dirname
(path)[source]¶ Return absolute path for the directory containing path
>>> abs_dirname('test/some.py') '/home/john/devel/test'
-
commodity.path.
child_relpath
(path, start='.')[source]¶ Convert a path to relative to the current directory if possible
>>> os.getcwd() /home/user >>> child_relpath('/home/user/doc/last') 'doc/last' >>> child_relpath('/usr/share/doc') '/usr/share/doc'
-
commodity.path.
find_in_ancestors
(fname, path)[source]¶ Search ‘fname’ file in the given ‘path’ and its ancestors
>>> find_in_ancestors('passwd', '/etc/network') '/etc' >>> find_in_ancestors('passwd', '/etc/network/interfaces') '/etc'
-
commodity.path.
resolve_path
(fname, paths, find_all=False)[source]¶ Search ‘fname’ in the given paths and return the first full path that has the file. If ‘find_all’ is True it returns all matching paths. It always returns a list.
>>> resolve_path('config', ['/home/user/brook', '/etc/brook']) ['/etc/brook/config']
-
commodity.path.
resolve_path_ancestors
(fname, path, find_all=False)[source]¶ Search ‘fname’ in the given path and its ancestors. It returns the first full path that has the file. If ‘find_all’ is True it returns all matching paths. Always returns a list
>>> resolve_path_ancestors('passwd', '/etc/network') '/etc/passwd' >>> resolve_path_ancestors('passwd', '/etc/network/interfaces') '/etc/passwd'
pattern¶
-
class
commodity.pattern.
Bunch
(*args, **kargs)[source]¶ It provides dict keys as attributes and viceversa
>>> data = dict(ccc=2) >>> bunch = Bunch(data) >>> bunch.ccc 2 >>> bunch.ddd = 3 >>> bunch['ddd'] 3 >>> bunch['eee'] = 4 >>> bunch.eee 4
-
class
commodity.pattern.
Flyweight
(name, bases, dct)[source]¶ Flyweight dessign pattern (for identical objects) as metaclass
>>> class Sample(object): ... __metaclass__ = Flyweight
-
class
commodity.pattern.
MetaBunch
(dct=None)[source]¶ A bunch of bunches. It allows to recursively access keys as attributes and viceversa. It may decorate any mapping type.
>>> b = MetaBunch() >>> b['aaa'] = {'bbb': 1} >>> b.aaa.bbb 1 >>> b.aaa.ccc = 2 >>> b['aaa']['ccc'] 2
-
class
commodity.pattern.
NestedBunch
(*args, **kargs)[source]¶ Bunch with recursive fallback in other bunch (its parent)
>>> a = NestedBunch() >>> a.foo = 1 >>> a2 = a.new_layer() >>> a2.bar = 2 >>> a2.foo 1 >>> a2.keys() ['foo', 'bar']
>>> b = NestedBunch() >>> b.foo = 1 >>> b2 = b.new_layer() >>> b2.foo 1 >>> b2['foo'] = 5000 >>> b2.foo 5000 >>> b['foo'] 1
>>> c = NestedBunch() >>> c.foo = 1 >>> c2 = c.new_layer().new_layer() >>> c2.foo 1
-
class
commodity.pattern.
TemplateBunch
[source]¶ A Bunch automatically templated with its own content
>>> t = TemplateBunch() >>> t.name = "Bob" >>> t.greeting = "Hi $name!" >>> t.greeting "Hi Bob!" >>> t.person = "$name's sister" >>> t.greeting = "Hi $person!" >>> t.person "Bob's sister" >>> t.greeting "Hi Bob's sister"
-
commodity.pattern.
memoized
(method_or_arg_spec=None, ignore=None)[source]¶ Memoized decorator for functions and methods, including classmethods, staticmethods and properties
>>> import random >>> >>> @memoized ... def get_identifier(obj): ... return random.randrange(1000) ... >>> n1 = get_identifier("hi") >>> n2 = get_identifier("hi") >>> assert n1 == n2
It allows to ignore some arguments in memoization, so different values for those arguments DO NOT implies new true invocations:
>>> @memoized(ignore=['b']) ... def gen_number(a, b): ... return random.randrange(1000) ... >>> n1 = gen_number(1, 2) >>> n2 = gen_number(1, 3) >>> print n1, n2 >>> assert n1 == n2
str¶
-
class
commodity.str_.
ASCII_TreeRender
[source]¶ Draws an ASCII-art tree just with ASCII characters. norm_brother = ‘+-‘
-
class
commodity.str_.
Printable
[source]¶ Class mixin that provides a __str__ method from the __unicode__ method.
-
class
commodity.str_.
TemplateFile
(template)[source]¶ Render a template (given as string or input file) and renders it with a dict over a string, given filename or tempfile.
>>> t = TemplateFile('$who likes $what') >>> t.render(dict(who='kate', what='rice')) 'kate likes rice'
testing¶
thread¶
-
class
commodity.thread_.
ActualReaderThread
(fdin, fdout)[source]¶ A thread that read from an actual file handler and write to an object that just has the write() method.
-
class
commodity.thread_.
SimpleThreadPool
(num_workers)[source]¶ A generic and simple thread pool. Function return value are received by means of callbacks.
>>> import math >>> >>> result = [] >>> pool = SimpleThreadPool(4) >>> pool.add(math.pow, (1, 2), callback=result.append) >>> pool.join()
Also implements a parallel
map()
for an argument sequence for a function executing each on a different thread:>>> pool = SimpleThreadPool(4) >>> pool.map(math.sqrt, ((2, 4, 5, 9))) (1.4142135623730951, 2.0, 2.23606797749979, 3.0)
>>> pool.map(math.pow, [2, 3, 3], [2, 2, 4]) (4, 9, 81)
-
class
commodity.thread_.
ThreadFunc
(target, *args, **kargs)[source]¶ Execute given function in a new thread. It provides return value (or exception) as object atribbutes.
>>> import math
>>> tf = ThreadFunc(math.sin, 8) >>> tf.join() >>> tf.result 0.9893582466233818
>>> tf = ThreadFunc(math.sqrt, -1) >>> tf.join() >>> tf.exception >>> ValueError('math domain error')
-
class
commodity.thread_.
WorkerGroup
(num_workers)[source]¶ A group of dedicated workers.
>>> import math >>> >>> results = [] >>> group = WorkerGroup(4) >>> w1 = group.get_worker("some unique value") >>> w1.add(math.square, (9,), callback=results.append) >>> group.join() >>> >>> print results [81]
-
commodity.thread_.
start_new_thread
(target, *args, **kargs)[source]¶ Execute given function in a new thread. It returns a
ThreadFunc
object.