Modules¶
wheezy.core¶
wheezy.core.benchmark¶
bechmark
module.
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 atindex
from the list. Ifkey
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 # doctest: +ELLIPSIS 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 = ['Hello World'.encode('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.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 inoptions
tries to get it frommaster
.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
overridemaster
.>>> 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 ofConfig
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) # doctest: +ELLIPSIS 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 bedatetime.timedelta
.>>> total_seconds(timedelta(hours=2)) 7200
or int:
>>> total_seconds(100) 100
otherwise raise
TypeError
.>>> total_seconds('100') # doctest: +ELLIPSIS Traceback (most recent call last): ... TypeError: ...
wheezy.core.db¶
session
module.
-
class
wheezy.core.db.
NullTPCSession
[source]¶ Null TPC session is supposed to be used in mock scenarios.
-
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).
-
connection
¶ Return the session connection. Not intended to be used directly, use cursor method instead.
-
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.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_go
(path=None, method='GET', params=None, headers=None, content_type='', body='')[source]¶ Sends HTTP AJAX request to web server.
-
content
¶ Returns a content of the response.
-
go
(path=None, method='GET', params=None, headers=None, content_type='', body='')[source]¶ Sends HTTP request to web server.
The
content_type
takes priority overparams
to usebody
. Thebody
can be a string or file like object.
-
json
¶ Returns a json response.
-
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()) # doctest: +ELLIPSIS [...]
Assess by language:
>>> lang = tm['en'] >>> m = lang['messages'] >>> m.gettext('hello') # doctest: +ELLIPSIS '...ello' >>> lang = tm['de'] >>> m = lang['messages'] >>> m.gettext('hello') # doctest: +ELLIPSIS '...'
Assess by translation domain:
>>> messages = tm.domains['messages'] >>> m = messages['en'] >>> m.gettext('hello') # doctest: +ELLIPSIS '...ello'
Fallback to English:
>>> m.gettext('world') # doctest: +ELLIPSIS '...orld'
If translation is unknown
key
returned>>> m.gettext('test') 'test'
-
wheezy.core.introspection¶
-
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
wheezy.core.json¶
json
module.
-
class
wheezy.core.json.
JSONEncoder
(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]¶ -
default
(obj)[source]¶ Implement this method in a subclass such that it returns a serializable object for
o
, or calls the base implementation (to raise aTypeError
).For example, to support arbitrary iterators, you could implement default like this:
def default(self, o): try: iterable = iter(o) except TypeError: pass else: return list(iterable) # Let the base class default method raise the TypeError return JSONEncoder.default(self, o)
-
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.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.
-
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.
wheezy.core.pooling¶
pooling
module.
-
class
wheezy.core.pooling.
EagerPool
(create_factory, size)[source]¶ Eager pool implementation.
Allocates all pool items during initialization.
-
count
¶ Returns a number of available items in the pool.
-
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 perslope
andstep
untilend
. A last delay is a time remaining. A total retry time is limited bytimeout
.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
andstep
- 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 urllib.parse 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 fromother
. Returns newUrlParts
instance.Query and Fragment parts are taken unconditionally from
other
.>>> from urllib.parse 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 instanceUrlParts
with partial content.parts
must be a 5-tuple: (scheme, netloc, path, query, fragment)>>> from urllib.parse 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', )) # doctest: +ELLIPSIS 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'