Modules

wheezy.core

wheezy.core.benchmark

bechmark module.

class wheezy.core.benchmark.Benchmark(targets, number, warmup_number=None, timer=None)[source]

Measure execution time of your code.

class wheezy.core.benchmark.Timer(target, name)[source]

Intercept a call to given method in order to compute timing.

wheezy.core.collections

The collections module contains types and functions that define various collections and algorithms.

class wheezy.core.collections.ItemAdapter(adaptee, index)[source]

Adapts defaultdict(list).__getitem__ accessor to return item at index from the list. If key is not found return None.

get(key, default=None)[source]

Return the value for key if key is in the adaptee, else default. If default is not given, it defaults to None, so that this method never raises a KeyError.

>>> d = defaultdict(list)
>>> d['a'].extend([1, 2, 3])
>>> a = ItemAdapter(d, 0)
>>> a.get('a')
1
>>> a.get('b', 100)
100
class wheezy.core.collections.attrdict[source]

A dictionary with attribute-style access. Maps attribute access to dictionary.

>>> d = attrdict(a=1, b=2)
>>> sorted(d.items())
[('a', 1), ('b', 2)]
>>> d.a
1
>>> d.c = 3
>>> d.c
3
>>> d.d 
Traceback (most recent call last):
    ...
AttributeError: ...
class wheezy.core.collections.defaultattrdict[source]

A dictionary with attribute-style access. Maps attribute access to dictionary.

>>> d = defaultattrdict(str, a=1, b=2)
>>> d.a
1
>>> d.c = 3
>>> d.c
3
>>> d.d
''
wheezy.core.collections.distinct(seq)[source]

Returns generator for unique items in seq with preserved order.

>>> list(distinct('1234512345'))
['1', '2', '3', '4', '5']

If the order is not important consider using set which is approximately eight times faster on large sequences.

>>> sorted(list(set('1234512345')))
['1', '2', '3', '4', '5']
wheezy.core.collections.first_item_adapter(adaptee)[source]

Adapts defaultdict(list).__getitem__ accessor to return the first item from the list.

>>> d = defaultdict(list)
>>> d['a'].extend([1, 2, 3])
>>> a = first_item_adapter(d)

Return a first item from the list.

>>> a['a']
1
wheezy.core.collections.gzip_iterator(items, compress_level=6)[source]

Iterates over items and returns generator of gzipped items.

items - a list of bytes

>>> items = [ntob('Hello World', 'latin1')]
>>> result = list(gzip_iterator(items))
>>> assert 3 == len(result)
>>> assert GZIP_HEADER == result[0]
wheezy.core.collections.last_item_adapter(adaptee)[source]

Adapts defaultdict(list).__getitem__ accessor to return the last item from the list.

>>> d = defaultdict(list)
>>> d['a'].extend([1, 2, 3])
>>> a = last_item_adapter(d)

Return a last item from the list.

>>> a['a']
3
wheezy.core.collections.list_values(keys, dictionary)[source]

Returns dictionary values orderd by keys.

>>> d = {'1': 1, '2': 2}
>>> list_values(['1', '2', '3'], d)
[1, 2, None]
wheezy.core.collections.map_keys(function, dictionary)[source]

Apply function to every key of dictionary and return a dictionary of the results.

>>> d = {'1': 1, '2': 2}
>>> sorted_items(map_keys(lambda key: 'k' + key, d))
[('k1', 1), ('k2', 2)]
wheezy.core.collections.map_values(function, dictionary)[source]

Apply function to every value of dictionary and return a dictionary of the results.

>>> d = {'1': 1, '2': 2}
>>> sorted_items(map_values(lambda value: 2 * value, d))
[('1', 2), ('2', 4)]
wheezy.core.collections.sorted_items(dictionary)[source]

Returns dictionary items sorted by key.

>>> d = {'1': 1, '2': 2}
>>> sorted_items(d)
[('1', 1), ('2', 2)]

wheezy.core.config

config module.

class wheezy.core.config.Config(options=None, master=None)[source]

Promotes options dict to attributes. If an option can not be found in options tries to get it from master. master must have a requested option otherwise raises error.

master can be a module.

>>> from sys import modules
>>> m = modules[Config.__module__]
>>> c = Config(master=m)
>>> c.DEBUG
False

or an instance of dict.

>>> c = Config(master={'DEBUG': False})
>>> c.DEBUG
False

options override master.

>>> c = Config(options={'DEBUG': True}, master=m)
>>> c.DEBUG
True

If option is not defined it takes from master.

>>> c = Config(master=m)
>>> c.DEBUG
False

Configs can be nested

>>> m = Config(dict(B='b'))
>>> c = Config(dict(A='a'), master=m)
>>> c.B
'b'

if options is an instance of Config than use its options only so this config can have own master.

>>> options = Config(dict(A='a'))
>>> c = Config(options)
>>> c.A
'a'

wheezy.core.datetime

datetime module.

wheezy.core.datetime.format_http_datetime(stamp)[source]

Formats datetime to a string following rfc1123 pattern.

>>> now = datetime(2011, 9, 19, 10, 45, 30, 0, UTC)
>>> format_http_datetime(now)
'Mon, 19 Sep 2011 10:45:30 GMT'

if timezone is not set in datetime instance the stamp is assumed to be in UTC (datetime.utcnow).

>>> now = datetime(2011, 9, 19, 10, 45, 30, 0)
>>> format_http_datetime(now)
'Mon, 19 Sep 2011 10:45:30 GMT'
>>> now = datetime.utcnow()
>>> assert format_http_datetime(now)

if stamp is a string just return it

>>> format_http_datetime('x')
'x'
>>> format_http_datetime(100) 
Traceback (most recent call last):
    ...
TypeError: ...
wheezy.core.datetime.format_iso_datetime(stamp)[source]

Return a string representing the date and time in ISO 8601 format. If the time is in UTC, adds a ‘Z’ directly after the time without a space.

see http://en.wikipedia.org/wiki/ISO_8601.

>>> class EET(tzinfo):
...     def utcoffset(self, dt):
...         return timedelta(minutes=120)
...     def dst(self, dt):
...         return timedelta()
>>> format_iso_datetime(datetime(2012, 2, 22, 12, 52, 29, 300))
'2012-02-22T12:52:29'
>>> format_iso_datetime(datetime(2012, 2, 22, 12, 52, 29, 300,
...     tzinfo=UTC))
'2012-02-22T12:52:29Z'
>>> format_iso_datetime(datetime(2012, 2, 22, 12, 52, 29, 300,
...     tzinfo=EET()))
'2012-02-22T12:52:29+02:00'
wheezy.core.datetime.format_iso_time(stamp)[source]

Return a string representing the time in ISO 8601 format. If the time is in UTC, adds a ‘Z’ directly after the time without a space.

see http://en.wikipedia.org/wiki/ISO_8601.

>>> class EET(tzinfo):
...     def utcoffset(self, dt):
...         return timedelta(minutes=120)
...     def dst(self, dt):
...         return timedelta()
>>> format_iso_time(time(12, 52, 29, 300))
'12:52:29'
>>> format_iso_time(time(12, 52, 29, 300,
...     tzinfo=UTC))
'12:52:29Z'
>>> format_iso_time(time(12, 52, 29, 300,
...     tzinfo=EET()))
'12:52:29+02:00'
wheezy.core.datetime.parse_http_datetime(stamp)[source]

Parses a string in rfc1123 format to datetime.

>>> parse_http_datetime('Mon, 19 Sep 2011 10:45:30 GMT')
datetime.datetime(2011, 9, 19, 10, 45, 30)
wheezy.core.datetime.total_seconds(delta)[source]

Returns a total number of seconds for the given delta.

delta can be datetime.timedelta.

>>> total_seconds(timedelta(hours=2))
7200

or int:

>>> total_seconds(100)
100

otherwise raise TypeError.

>>> total_seconds('100') 
Traceback (most recent call last):
    ...
TypeError: ...
class wheezy.core.datetime.utc(name)[source]

UTC timezone.

dst(dt)[source]

DST is not in effect.

>>> UTC.dst(None)
datetime.timedelta(0)
tzname(dt)[source]

Name of time zone.

>>> GMT.tzname(None)
'GMT'
>>> UTC.tzname(None)
'UTC'
utcoffset(dt)[source]

Offset from UTC.

>>> UTC.utcoffset(None)
datetime.timedelta(0)

wheezy.core.db

session module.

class wheezy.core.db.NullSession[source]

Null session is supposed to be used in mock scenarios.

commit()[source]

Simulates commit. Asserts the session is used in scope.

cursor(*args, **kwargs)[source]

Ensure session is entered.

class wheezy.core.db.NullTPCSession[source]

Null TPC session is supposed to be used in mock scenarios.

commit()[source]

Simulates commit. Asserts the session is used in scope.

enlist(session)[source]

Ensure session is entered.

class wheezy.core.db.Session(pool)[source]

Session works with a pool of database connections. Database connection must be implemented per Database API Specification v2.0 (see PEP0249).

commit()[source]

Commit any pending transaction to the database.

connection None[source]

Return the session connection. Not intended to be used directly, use cursor method instead.

cursor(*args, **kwargs)[source]

Return a new cursor object using the session connection.

class wheezy.core.db.TPCSession(format_id=7, global_transaction_id=None, branch_qualifier='')[source]

Two-Phase Commit protocol session that works with a pool of database connections. Database connection must be implemented per Database API Specification v2.0 (see PEP0249).

commit()[source]

Commit any pending transaction to the database.

enlist(session)[source]

Begins a TPC transaction with the given session.

wheezy.core.descriptors

descriptors module.

class wheezy.core.descriptors.attribute(f)[source]

attribute decorator is intended to promote a function call to object attribute. This means the function is called once and replaced with returned value.

>>> class A:
...     def __init__(self):
...         self.counter = 0
...     @attribute
...     def count(self):
...         self.counter += 1
...         return self.counter
>>> a = A()
>>> a.count
1
>>> a.count
1

wheezy.core.feistel

feistel module.

wheezy.core.feistel.make_feistel_number(f)[source]

Generate pseudo random consistent reversal number per Feistel cypher algorithm.

see http://en.wikipedia.org/wiki/Feistel_cipher

>>> feistel_number = make_feistel_number(sample_f)
>>> feistel_number(1)
573852158
>>> feistel_number(2)
1788827948
>>> feistel_number(123456789)
1466105040

Reversable

>>> feistel_number(1466105040)
123456789
>>> feistel_number(1788827948)
2
>>> feistel_number(573852158)
1

wheezy.core.gzip

One-shot compression and decompression.

wheezy.core.gzip.compress(data, compresslevel=9)[source]

Compress data in one shot.

wheezy.core.gzip.decompress(data)[source]

Decompress data in one shot.

wheezy.core.httpclient

class wheezy.core.httpclient.HTTPClient(url, headers=None)[source]

HTTP client sends HTTP requests to server in order to accomplish an application specific use cases, e.g. remote web server API, etc.

ajax_get(path, **kwargs)[source]

Sends GET HTTP AJAX request.

ajax_go(path=None, method='GET', params=None, headers=None, content_type='', body='')[source]

Sends HTTP AJAX request to web server.

ajax_post(path, **kwargs)[source]

Sends POST HTTP AJAX request.

content None[source]

Returns a content of the response.

follow()[source]

Follows HTTP redirect (e.g. status code 302).

get(path, **kwargs)[source]

Sends GET HTTP request.

go(path=None, method='GET', params=None, headers=None, content_type='', body='')[source]

Sends HTTP request to web server.

The content_type takes priority over params to use body. The body can be a string or file like object.

head(path, **kwargs)[source]

Sends HEAD HTTP request.

json None[source]

Returns a json response.

post(path, **kwargs)[source]

Sends POST HTTP request.

wheezy.core.i18n

i18n module.

class wheezy.core.i18n.TranslationsManager(directories=None, default_lang='en')[source]

Manages several languages and translation domains.

add_fallback(languages)[source]

Adds fallback languages.

>>> tm = TranslationsManager()
>>> tm.add_fallback(('uk', 'ru'))
>>> tm.fallbacks
{'uk': ('uk', 'ru', 'en')}
load(localedir)[source]

Load all available languages and domains from the given directory.

{localedir}/{lang}/LC_MESSAGES/{domain}.mo

In order to generate .mo file from .po file: $ msgfmt domain.po

>>> curdir = os.path.dirname(__file__)
>>> localedir = os.path.join(curdir, 'tests', 'i18n')
>>> tm = TranslationsManager()
>>> tm.load(localedir)
>>> sorted(tm.translations.keys())
['de', 'en']

Assess by language:

>>> lang = tm['en']
>>> m = lang['messages']
>>> m.gettext('hello')
'Hello'
>>> lang = tm['de']
>>> m = lang['messages']
>>> m.gettext('hello')
'Hallo'

Assess by translation domain:

>>> messages = tm.domains['messages']
>>> m = messages['en']
>>> m.gettext('hello')
'Hello'

Fallback to English:

>>> m.gettext('world')
'World'

If translation is unknown key returned

>>> m.gettext('test')
'test'

wheezy.core.introspection

introspection module.

wheezy.core.introspection.import_name(fullname)[source]

Dynamically imports object by its full name.

>>> from datetime import timedelta
>>> import_name('datetime.timedelta') is timedelta
True
class wheezy.core.introspection.looks(cls)[source]

Performs duck typing checks for two classes.

Typical use:

assert looks(IFoo, ignore_argspec=['pex']).like(Foo)
like(cls, notice=None, ignore_funcs=None, ignore_argspec=None)[source]

Check if self.cls can be used as duck typing for cls.

cls - class to be checked for duck typing. ignore_funcs - a list of functions to ignore ignore_argspec - a list of functions to ignore arguments spec.

wheezy.core.json

json module.

wheezy.core.json.json_decode(s)[source]

Decode s (containing a JSON unicode string) to a Python object.

wheezy.core.json.json_encode(obj)[source]

Encode obj as a JSON formatted unicode string.

Correctly escapes forward slash to be able embed javascript code.

wheezy.core.luhn

luhn module.

wheezy.core.luhn.digits_of(n)[source]

Returns a list of all digits from given number.

>>> digits_of(123456789)
[1, 2, 3, 4, 5, 6, 7, 8, 9]
wheezy.core.luhn.is_luhn_valid(n)[source]
>>> is_luhn_valid(1234567897)
True
>>> is_luhn_valid(473802106)
True
>>> is_luhn_valid(34518893)
False
wheezy.core.luhn.luhn_checksum(n)[source]

Calculates checksum based on Luhn algorithm, also known as the “modulus 10” algorithm.

see http://en.wikipedia.org/wiki/Luhn_algorithm

>>> luhn_checksum(1788827948)
0
>>> luhn_checksum(573852158)
1
>>> luhn_checksum(123456789)
7
wheezy.core.luhn.luhn_sign(n)[source]

Signs given number by Luhn checksum.

>>> luhn_sign(78482748)
784827487
>>> luhn_sign(47380210)
473802106
>>> luhn_sign(123456789)
1234567897
wheezy.core.luhn.sum2digits(d)[source]

Sum digits of a number that is less or equal 18.

>>> sum2digits(2)
2
>>> sum2digits(17)
8

wheezy.core.mail

mail module.

class wheezy.core.mail.Alternative(content, content_type='text/html', charset=None)[source]

Represents alternative mail message.

class wheezy.core.mail.Attachment(name, content, content_type=None, disposition=None, name_charset=None, content_charset=None)[source]

An attachment to mail message.

classmethod from_file(path)[source]

Creates an attachment from file.

class wheezy.core.mail.MailMessage(subject='', content='', from_addr=None, to_addrs=None, cc_addrs=None, bcc_addrs=None, reply_to_addrs=None, content_type='text/plain', charset='us-ascii')[source]

Mail message.

class wheezy.core.mail.Related(content_id, content, content_type)[source]

A resource related to alternative mail message.

classmethod from_file(path)[source]

Creates a related mail resource from file.

class wheezy.core.mail.SMTPClient(host='127.0.0.1', port=25, use_tls=False, username=None, password=None)[source]

SMTP client that can be used to send mail.

send(message)[source]

Sends a single mail message.

send_multi(messages)[source]

Sends multiple mail messages.

wheezy.core.mail.mail_address(addr, name=None, charset='utf8')[source]

Returns mail address formatted string.

wheezy.core.pooling

pooling module.

class wheezy.core.pooling.EagerPool(create_factory, size)[source]

Eager pool implementation.

Allocates all pool items during initialization.

count None[source]

Returns a number of available items in the pool.

class wheezy.core.pooling.LazyPool(create_factory, size)[source]

Lazy pool implementation.

Allocates pool items as necessary.

acquire()[source]

Return an item from pool. Blocks until an item is available.

count None[source]

Returns a number of available items in the pool.

class wheezy.core.pooling.Pooled(pool)[source]

Pooled serves context manager purpose, effectively acquiring and returning item to the pool.

Here is an example:

with Pooled(pool) as item:
    # do something with item

wheezy.core.retry

retry module.

wheezy.core.retry.make_retry(timeout, start, end=None, slope=1.0, step=0.0)[source]

Return a function that accepts a single argument acquire which should be a callable (without any arguments) that returns a boolean value when attempt to acquire some resource or perform operation succeeded or not.

If a first attempt fails the retry function sleeps for start and later delay time is increased linearly per slope and step until end. A last delay is a time remaining. A total retry time is limited by timeout.

timeout - a number of seconds for the entire retry operation from the first failed attempt to the last (excluding time for both acquire operations).

start - a time for initial delay.

end - a time for a longest delay between retry attempts.

slope and step - coefficients for linear calculation of the next delay.

Example 1:

# delays: 0.5, 0.5, 0.5, 0.5
retry = make_retry(timeout=10.0, start=0.5)

Example 2:

# delays: 0.5, 0.1, 1.5, 2.0
retry = make_retry(timeout=10.0, start=0.5, end=2.0, step=0.5)

Example 3:

# delays: 0.05, 0.1, 0.2, 0.4, 0.8, 1.6, 2.0
retry = make_retry(timeout=10.0, start=0.05, end=2.0, slope=2.0)
if retry(lambda: acquire('something')):
    # good to go
else:
    # timed out

wheezy.core.url

url module

class wheezy.core.url.UrlParts(parts)[source]

Concrete class for urlparse.urlsplit() results.

geturl()[source]

Return the re-combined version of the original URL as a string.

>>> from wheezy.core.comp import urlsplit
>>> parts = urlsplit('http://www.python.org/dev/peps/pep-3333')
>>> parts = urlparts(parts)
>>> parts.geturl()
'http://www.python.org/dev/peps/pep-3333'
join(other)[source]

Joins with another UrlParts instance by taking none-empty values from other. Returns new UrlParts instance.

Query and Fragment parts are taken unconditionally from other.

>>> from wheezy.core.comp import urlsplit
>>> parts = urlsplit('http://www.python.org/dev/peps/pep-3333')
>>> parts = urlparts(parts)
>>> parts = parts.join(urlparts(scheme='https', path='/test'))
>>> parts.geturl()
'https://www.python.org/test'
wheezy.core.url.urlparts(parts=None, scheme=None, netloc=None, path=None, query=None, fragment=None)[source]

Factory function for UrlParts that create an instance UrlParts with partial content.

parts must be a 5-tuple: (scheme, netloc, path, query, fragment)

>>> from wheezy.core.comp import urlsplit
>>> parts = urlsplit('http://www.python.org/dev/peps/pep-3333')
>>> urlparts(parts)
urlparts('http', 'www.python.org', '/dev/peps/pep-3333', '', '')
>>> urlparts(scheme='https', path='/test')
urlparts('https', None, '/test', None, None)

Otherwise raise assertion error

>>> urlparts(('https', )) 
Traceback (most recent call last):
    ...
AssertionError: ...

wheezy.core.uuid

uuid module.

wheezy.core.uuid.parse_uuid(s)[source]
>>> parse_uuid('pK8vVOmIT1y_1jUceSmbdA')
UUID('a4af2f54-e988-4f5c-bfd6-351c79299b74')
>>> parse_uuid('0Xq6iBnDQA6t7j7Pk12ycg')
UUID('d17aba88-19c3-400e-adee-3ecf935db272')
>>> parse_uuid('Oa4T7iAqQtGRF2-2_dFppA')
UUID('39ae13ee-202a-42d1-9117-6fb6fdd169a4')
>>> parse_uuid('AAAAAAAAAAAAAAAAAAAAAA')
UUID('00000000-0000-0000-0000-000000000000')

Return an empty uuid in case the string is empty or length not equal to 22.

>>> parse_uuid('')
UUID('00000000-0000-0000-0000-000000000000')
>>> parse_uuid('x')
UUID('00000000-0000-0000-0000-000000000000')

Incorrect base64 padding.

>>> parse_uuid('AAAAAAAAAA*AAAAAAAAAAA')
UUID('00000000-0000-0000-0000-000000000000')
wheezy.core.uuid.shrink_uuid(uuid)[source]

Returns base64 representation of uuid.

>>> shrink_uuid(UUID('a4af2f54-e988-4f5c-bfd6-351c79299b74'))
'pK8vVOmIT1y_1jUceSmbdA'
>>> shrink_uuid(UUID('d17aba88-19c3-400e-adee-3ecf935db272'))
'0Xq6iBnDQA6t7j7Pk12ycg'
>>> shrink_uuid(UUID('39ae13ee-202a-42d1-9117-6fb6fdd169a4'))
'Oa4T7iAqQtGRF2-2_dFppA'
>>> shrink_uuid(UUID_EMPTY)
'AAAAAAAAAAAAAAAAAAAAAA'