diff options
| author | Eli Collins <elic@assurancetechnologies.com> | 2011-01-06 01:15:05 +0000 |
|---|---|---|
| committer | Eli Collins <elic@assurancetechnologies.com> | 2011-01-06 01:15:05 +0000 |
| commit | 1d2669f6f50d3cedc4ffd5439780b13492469bdd (patch) | |
| tree | c56c9a0883aee5be78624920b85cb6e155779049 /bps/misc.py | |
| parent | 0dd599bbb323387991cd8a3565ea87f36ff0892f (diff) | |
| download | passlib-1d2669f6f50d3cedc4ffd5439780b13492469bdd.tar.gz | |
tweaked config files, moved dir to passlib
Diffstat (limited to 'bps/misc.py')
| -rw-r--r-- | bps/misc.py | 487 |
1 files changed, 0 insertions, 487 deletions
diff --git a/bps/misc.py b/bps/misc.py deleted file mode 100644 index e81e44c..0000000 --- a/bps/misc.py +++ /dev/null @@ -1,487 +0,0 @@ -"""bps.misc -- assorted functions that dont fit in another category""" -#=================================================== -#imports -#=================================================== -#core -from functools import update_wrapper -import time -import re -#pkg -from bps.types import Undef -from bps.meta import find_attribute -#local -__all__ = [ - #property constructors - 'indirect_property', - 'constructor_property', - - #http - 'parse_agent_string', - 'agent_string_has_product', - - #other - 'stepped_delay', -] - -#========================================================= -#property constructors -#========================================================= -class indirect_property(object): - """descriptor which acts like property(), but resolves methods at instance time. - - One of the drawbacks of the builtin :func:``property`` is that it stored - the functions directly. Thus, if a subclass overrides the method - which is also being used by a property's fget, - the property object will still use the original function. - - This is a drop-in replacement for property which takes in - attribute names instead of actual functions. It does - runtime resolution of the attributes, so that the named - methods can be safely overridden (even on a per-instance basis) - and still have the properties use the correct code. - - .. note:: - Due to the repeated additional lookup, this is slower - than a normal property, so use it only if you have to. - """ - #TODO: need to make this work right for various border cases (missing fget/fset) - #TODO: default doc based on attr names - - def __init__(self, fget=None, fset=None, fdel=None, doc=None): - self.fget = fget - self.fset = fset - self.fdel = fdel - if doc: - self.__doc__ = doc - - def __get__(self, obj, cls): - if obj is None: - return self - else: - return getattr(obj, self.fget)() - - def __set__(self, obj, value): - if self.fset: - getattr(obj, self.fset)(value) - else: - raise AttributeError("readonly attribute") - - def __delete__(self, obj): - if self.fdel: - getattr(obj, self.fdel)(value) - else: - raise AttributeError("can't delete attribute") - -class constructor_property(object): - """lazy-initialized attribute. - - This is a class property, - which takes in a constructor func, and uses that function - to fill in the instance attribute when it's first accessed. - - usage:: - >>> from bps.misc import constructor_property - >>> #create a custom class - >>> class Example(object): - >>> a = constructor_property(dict) - >>> e = Example() - >>> #initially nothing is stored in 'a' - >>> e.__dict__.get("a") - None - >>> #but when it's first accessed, dict() is called, and the value is stored/returned - >>> e.a - {} - >>> #from then on, that's the value that will be returned for .a, until ``del e.a`` is called - >>> e.__dict__.get("a") - {} - - :arg func: - function / class to call when attribute is first accessed for an instance - :arg name: - optionally let object know which attribute it's stored under - (will be autodiscovered later) - :param passref: - if True, func will be called with instance as first argument (eg ``func(self)``) - rather than without arguments (eg ``func()``) - """ - def __init__(self, func, name=None, passref=False): - self.func = func - self.name = name - self.passref = passref - - def __get__(self, obj, cls): - if obj is None: - return self - if self.name is None: - self.name = find_attribute(cls, self, required=True) - assert self.name not in obj.__dict__ - if self.passref: - value = self.func(obj) - else: - value = self.func() - obj.__dict__[self.name] = value - #we should never get called again for this object - return value - -class class_property(object): - "classmethod+readonly property" - #TODO: document this - def __init__(self, fget): - self.fget = fget - def __get__(self, owner, cls): - return self.fget(cls) - -#========================================================= -# -#========================================================= -def _iter_decay(lower, upper, half): - "helper for stepped_delay" - #default delay loop using "lower" "upper" and "half" - #equation: delay[idx] = upper - (upper-lower) * (decay ** idx) - #such that: - # delay[0] == lower - # delay[half] = (upper+lower)/2 - # delay[idx] < upper - # - #this means decay = (1/2)**(1/half) - # - if half: - decay = .5**(1.0/half) - else: - decay = .9 ## approx ~ half=7 - value = upper-lower - while True: - yield upper-value - value *= decay - -def stepped_delay(timeout=None, count=None, steps=None, lower=.1, upper=90, half=None): - """generate a stepped delay loop; useful when polling a resource repeatedly. - - This function provides a delay loop - for such things as polling a filesystem for changes, etc. - It provides an initially short delay which slowly backs off. - It's designed to be used an iterator, so that all logic - stays within your application. - - You can either specify a custom sequence of delay values via *steps*, - or use the default exponential decay algorithm, which - begans with a delay of *lower*, and slowly increases, - approaching a delay time of *upper*. - - :param timeout: - If specified, the loop will stop after *timeout* seconds - have passed, no matter how many repetitions have been run. - - :param count: - If specified, the loop will stop after *count* repetitions. - - :param steps: - If specified, this should be a sequence - of delay values to use. When the sequence runs - out, the last delay value will be repeated. - If *steps* is not used, a default exponential - decay algorithm will be used. - - :param lower: - [ignored if *steps* is specified] - This specifies the starting delay. - The first delay will be this length, - the next a little more, and so on. - - :param upper: - [ignored if *steps* is specified] - This specifies the upper bound on the delay. - Each time the iterator sleeps, the delay - will increase, asymptotically approaching - the *upper* bound. - - :param half: - [optional, ignored if *steps* is specified] - If specified, adjusts the rate of the exponential delay - increase such that it will take exactly *half* - rounds through the iterator before the delay - is at the half-way mark between *lower* and *upper*. - - :Returns: - This loop yields tuples of ``(index,delay)``, - where *index* is the number of passes that have been made, - and *delay* is the amount of time it slept before - yielding the last tuple. It will increase the delay - used each time before it yeilds a new tuple, - in accordance with the configuration above. - If the loop ends due to *timeout* or *count*, - the iterator will raise :exc:`StopIteration`. - - Usage Example:: - - >>> import time - >>> from bps.misc import stepped_delay - >>> for i,d in stepped_delay(count=10, lower=.1, upper=10): - >>> print i,d,time.time() - >>> #... do stuff, calling break if done with loop - >>> else: - >>> print "loop exit w/o success" - 0 0 1244648293.01 - 1 0.1 1244648293.11 - 2 1.09 1244648294.2 - 3 1.981 1244648296.19 - 4 2.7829 1244648298.97 - 5 3.50461 1244648302.48 - 6 4.154149 1244648306.64 - 7 4.7387341 1244648311.38 - 8 5.26486069 1244648316.65 - 9 5.738374621 1244648322.39 - loop exit w/o success - - .. todo:: - Could allow delay to be reset to initial value - by sending ``"reset"`` back to the yield statement. - """ - - #run first round without any delay - yield 0, 0 - - #prepare delay value generator - if steps: - #ignore 'lower', 'upper', and 'half' - def loopgen(): - for value in steps: - yield value - while True: #repeat last value - yield value - loop = loopgen() - else: - if upper <= lower: #allow us to set really small 'upper' and auto-scale lower - lower = .1 * upper - loop = _iter_decay(lower, upper, half) - - #run main delay loop - if timeout: - end = time.time() + timeout - for idx, delay in enumerate(loop): - time.sleep(delay) - yield idx+1, delay - #check if it's time to abort - if count and idx+2 >= count: - return - if timeout and time.time() >= end: - return - -#========================================================= -#http agent string -#========================================================= -_clean_re = re.compile(r"\s+") - -_agent_re = re.compile( - r""" - ^ - \s* - ( - (?P<product> - (?P<name> - [^\s()/]+ # technically this should only be TOKEN chars - ) - ( - / - (?P<version> - [^\s()]+ #XXX: what _is_ allowed here? TOKEN? - ) - )? - ) - | - ( - \( - (?P<comment> - [^)]+ #technically this should only be TOKEN chars - ) - \) - ) - ) - \s* - """, re.I|re.X) - -def parse_agent_string(value, normalize=True): - """parse a HTTP user agent string. - - This parses an HTTP User Agent string, - returning a list of agents identified in the string, in order. - - - :type value: str - :param value: - The agent string to parse - - :type normalize: bool - :param normalize: - This flag (enabled by default) - turns on any special-case heuristics for known - atypical user agent strings, as well - as converting the string to lower case. - It can be disabled to get the unmangled results. - - :returns: - A list of dictionaries, one for each product found. - The first dictionary is usually considered the primary. - This code assumes comments will always *follow* the product description - they are attached to, but if this rule is violated, - a "blank" product entry will be inserted, where all relevant keys - except "comment" will be ``None``. Other than that case, - the following keys should be filled out for each dictionary: - - product - This will contain the raw product name, eg "Mozilla/5.0". - name - This will contain just the name of the product - (assuming it has the format "name/version"). - If the product couldn't be parsed this way, name's contents are undefined. - version - This will contain the version of the product, - (assuming it has the format "name/version"). - If the product couldn't be parsed this way, version's contents are undefined. - comment - This is present if a comment stanza followed - the product definition. This will be a list of strings, - as read from the comment and separated by semicolons. - If no comment is present, the key will not be included. - - Usage Example:: - - >>> from bps.misc import parse_agent_string - >>> parse_agent_string("Mozilla/5.0 (X11; U; Linux x86_64; en-US; rv:1.9.0.11) Gecko/2009060309 Ubuntu/9.04 (jaunty) Firefox/3.0.11") - [ - { 'name': 'Mozilla', 'version': '5.0', - 'product': 'Mozilla/5.0', - 'comments': ['X11', 'U', 'Linux x86_64', - 'en-US', 'rv:1.9.0.11'], - }, - { 'name': 'Gecko', 'product': 'Gecko/2009060309', - 'version': '2009060309' - }, - { 'name': 'Ubuntu', 'version': '9.04', - 'product': 'Ubuntu/9.04', - 'comments': ['jaunty'], - }, - { 'name': 'Firefox', 'version': '3.0.11', - 'product': 'Firefox/3.0.11', - } - ] - - .. seealso: - - :rfc:`2068` is the authoritative agent string format spec. - """ - #NOTE: this code makes the assumption - #that a comment will always be FOLLOWING (and is associated with) the preceding product. - #this goes against the grammar of RFC2068, but is the de facto case. - #thus, if a unexpected comment is encountered, a empty product entry will be created. - orig = value - value = _clean_re.sub(" ", value).strip() - if normalize: - value = value.lower() - out = [] - while value: - m = _agent_re.match(value) - if m: - comment = m.group("comment") - if comment: - comments = [ elem.strip() for elem in comment.split(";") ] - if out and isinstance(out[-1], dict) and 'comments' not in out[-1]: - out[-1]['comments'] = comments - else: - log.warning("unexpected comment segment in agent: %r %r", comment, orig) - out.append(dict(product=None, name=None, version=None, comments=comments)) - else: - product, name, version = m.group("product", "name", "version") - out.append(dict(product=product, name=name, version=version)) - value = value[m.end():] - else: - #can this _ever_ happen? - log.warning("failed to parse agent segment: %r of %r", value, orig) - value = '' -## if not normalize: -## return out - #TODO: detect the "+http://homepage" elements add end of comment list, - # move out to "url" kwd - #TODO: detect platform info - #TODO: detect firefox, opera, konq, safari, chrome, - # and move their products to the front -## #now we apply various bits of UA-specific knowledge to normalize things -## #TODO: could pull out 'MSIE' etc -## for entry in out: -## if not entry['product'] or not entry['comments']: -## continue -## #could parse out site urls - return out - -def _parse_agent_version(value): - if value is None: - return None - #XXX: use a real version parser here. - if isinstance(value, str): - try: - return tuple(int(x) for x in value.split(".")) - except ValueError: - return None - elif isinstance(value, int): - return tuple(value) - #should be tuple of ints. - return value - -def agent_string_has_product(agent, name, min_version=None): - """tests if agent string references a product name. - - This wrapper for :func:`parse_agent_string` - checks if a given product is found in the provided string. - This is a simple function, more complex cases may require - rolling your own test function. - - :param agent: - The raw agent string, OR the output of parse_agent_string. - :param name: - The name of the product to check for. - :param min_version: - Optional minimum version. - For this to work, min_version must be an integer, - tuple of integers, or a period-separated string. - - :returns: - Returns ``True`` if a match is found, - ``False`` if a match is not found. - """ - name = name.lower() - min_version = _parse_agent_version(min_version) - if isinstance(agent, str): - agent = parse_agent_string(agent) - for entry in agent: - if entry['name'] == name: - if not min_version or min_version <= _parse_agent_version(entry['version']): - return True - #TODO: IE detect here or in extended? - return False - -#========================================================= -#code scraps -#========================================================= - -#need to clean this up a little, but might be useful -##def formatFuncStr(fname, *args, **kwds): -## if isinstance(fname, str): -## pass## elif callable(fname): -## fname = fname.__name__ -## else: -## fname = str(fname) -## -## body = "" -## if args: -## for a in args: -## if body != "": body += "," -## body += repr(a) -## if kwds: -## for k,v in kwds.items(): -## if body != "": body += "," -## body += "%s=%r" % (k,v) -## return "%s(%s)" % (fname,body) - -#========================================================= -# -#========================================================= |
