DZone Snippets: decorator code

Pattern Matching based WSGI-enabled URL routing tool.

sevkin (Vsevolod Balashov) — Mon, 09 Jun 2008 07:10:57 GMT

Actual version on http://pypi.python.org/pypi/decoroute



easy_install decoroute

Example:



import decoroute

    

app = decoroute.App()

    

def render_response(status = '200 OK', **kw):

    # try your favorite templating engine here

    return status, [('Content-Type', 'text/plain')], [str(kw)]



def redirect_to(env, endpoint, **kw):

    return '302 FOUND', [('Content-Type', 'text/plain'), \

    ('Location', '%s://%s%s' % (env['wsgi.url_scheme'], env['HTTP_HOST'], \

    env['decoroute.app'].url_for(endpoint, **kw)))], ['']

    

@app.expose('/node', id = '1')

@app.expose('/node/')

def node(env, id):

    return render_response(id = id)

    

@app.expose('/url_for')

def url_for(env):

    return render_response( \

        url = env['decoroute.app'].url_for(node, id = 666))

    

@app.expose('/302')

def found(env):

    return redirect_to(env, not_found)

    

@app.expose('/404')

def not_found(env):

    raise decoroute.NotFound()

    

@app.not_found()

def not_found_handler(env):

    return render_response('404 NF', url = env['PATH_INFO'])

    

from wsgiref.simple_server import make_server

    

make_server('', 5555, app).serve_forever()

Complete source < 100 lines of code:



#!/usr/bin/env python

# vim:ts=4:sw=4:et

# -*- coding: utf-8 -*-

# $Id: decoroute.py,v 6eb37aeee5cd 2008/06/09 07:38:26 vsevolod $

#

# Pattern Matching based WSGI-enabled URL routing tool.

# Actual version on http://pypi.python.org/pypi/decoroute

# (C) 2008 by Vsevolod S. Balashov 

# under terms of GNU LGPL v2.1 http://www.gnu.org/licenses/old-licenses/lgpl-2.1.txt



__author__ = "Vsevolod Balashov"

__email__ = "vsevolod at balashov dot name"



import re

from sets import ImmutableSet

import wsgistraw



__all__ = ['NotFound', 'App']



class NotFound(Exception):

    pass



_pattern_re = re.compile(r'''

    ([^<]+)                     # static rule data

    (?:<

        ([a-zA-Z][a-zA-Z0-9_]*) # variable name

        \:

        ([^>]+)                 # regexp constraint

    >)?

''', re.VERBOSE)



def pattern2regexp(pattern, f, s = lambda x: re.escape(x)):

    def parser():

        for t in _pattern_re.findall(pattern):

            yield s(t[0])

            if t[1] != '':

                yield f(t[1], t[2])

    return parser()



make_url_for = lambda p: ''.join(pattern2regexp(p, lambda v, r: '%%(%s)s' % v, lambda s: s))

make_variables = lambda p: filter(lambda x: x, pattern2regexp(p, lambda v, r: v, lambda s: None))

make_pattern = lambda p: r''.join(pattern2regexp(p, lambda v, r: r'(?P<%s>%s)' % (v, r)))

make_selector_fragment = lambda p: r''.join(pattern2regexp(p, lambda v, r: r'(?:%s)' % r))

make_selector = lambda i: re.compile(r'(^%s$)' % r'$)|(^'.join(map(make_selector_fragment, i)))



class UrlMap(object):

    def __init__(self):

        self._endpoints = {}

        self._patterns = {}

        self._pattern_selector = make_selector(self._patterns.iterkeys())

    

    def add(self, pattern, endpoint, **kw):

        if self._patterns.has_key(pattern):

            raise Exception('duplicate pattern', pattern)

        self._endpoints[(endpoint, ImmutableSet(make_variables(pattern)))] = make_url_for(pattern)

        self._patterns[pattern] = (re.compile(make_pattern(pattern)), endpoint, kw)

        self._pattern_selector = make_selector(self._patterns.iterkeys())

    

    def route(self, url):

        try:

            p = self._patterns.values()[re.match(self._pattern_selector, url).lastindex - 1]

            d = re.match(p[0], url).groupdict().copy()

            d.update(p[2])

            return p[1], d

        except:

            raise NotFound('route not found', url)

    

    def url_for(self, endpoint, **kw):

        return self._endpoints[(endpoint, ImmutableSet(kw.keys()))] % kw



class App(object):

    def __init__(self, key = 'decoroute.app'):

        self._key = key

        self._map = UrlMap()

        self._not_found = lambda e: ('404 NOT FOUND', [("Content-Type", "text/plain")], [''])

    

    @wsgistraw.app

    def __call__(self, env):

        try:

            env[self._key] = self

            endpoint, kw = self._map.route(env['PATH_INFO'])

            return endpoint(env, **kw)

        except NotFound:

            return self._not_found(env)

    

    def expose(self, pattern, **kw):

        def decorate(f):

            self._map.add(pattern, f, **kw)

            return f

        return decorate

    

    def not_found(self):

        def decorate(f):

            self._not_found = f

            return f

        return decorate

    

    def url_for(self, endpoint, **kw):

        return self._map.url_for(endpoint, **kw)

Thread Pool and Threaded Python decorators

sevkin (Vsevolod Balashov) — Thu, 16 Aug 2007 12:55:37 GMT



#!/usr/bin/env python

# -*- coding: utf-8 -*-

# vim:ts=4:sw=4:et

# (c) 2007 under terms of LGPL v 2.1

# by Vsevolod S. Balashov  



from threading import Thread



def threaded(func):

    def proxy(*args, **kwargs):

        thread = Thread(target=func, args=args, kwargs=kwargs)

        thread.start()

        return thread

    return proxy



from Queue import Queue



class Pool(Queue):

    def __init__(self, maxsize):

        assert maxsize > 0, 'maxsize > 0 required for Pool class'

        Queue.__init__(self, maxsize)

        for i in range(maxsize):

            thread = Thread(target = self._worker)

            thread.setDaemon(True)

            thread.start()



    def _worker(self):

        while True:

            try:

                func, args, kwargs = self.get()

                func(*args, **kwargs)

            except:

                self.task_done()

                self.join()

                raise

            self.task_done()



    def addJob(self, func, *args, **kwargs):

        self.put((func, args, kwargs))



    def __enter__(self):

        pass



    def __exit__(self, exc_type, exc_value, traceback):

        self.join()



def threadpool(pool):

    assert pool.__class__ == Pool, 'threadpool decorator require a Pool object'

    def decorator(func):

        def proxy(*args, **kwargs):

            pool.put((func, args, kwargs))

            return pool

        return proxy

    return decorator

example usage



from threadpool import *



from time import sleep

from random import random



pool = Pool(3)

    

@threadpool(pool)

def test_threadpool(i):

    print 'threadpool %i enter' % i

    sleep(random())

    print 'threadpool %i exit' % i



print 'threadpool example'

for i in range(6):

    test_threadpool(i)

pool.join()

print 'done'

print ''



@threaded

def test_threaded(i):

    print 'threaded %i enter' % i

    sleep(random())

    print 'threaded %i exit' % i



print 'threaded example'

for i in range(5):

    test_threaded(i)

print 'done'

result output

threadpool example
threadpool 0 enter
threadpool 1 enter
threadpool 2 enter
threadpool 0 exit
threadpool 3 enter
threadpool 1 exit
threadpool 4 enter
threadpool 2 exit
threadpool 5 enter
threadpool 3 exit
threadpool 5 exit
threadpool 4 exit
done

threaded example
done
threaded 0 enter
threaded 1 enter
threaded 2 enter
threaded 3 enter
threaded 4 enter
threaded 1 exit
threaded 2 exit
threaded 3 exit
threaded 0 exit
threaded 4 exit

Python profile decorator

maxme (Maxime Biais) — Mon, 22 Jan 2007 11:03:10 GMT

Python profile decorator. More info on this blog post: http://www.biais.org/blog/index.php/2007/01/20/18-python-profiling-decorator



# Maxime Biais 



import hotshot, hotshot.stats

 

def profileit(printlines=1):

    def _my(func):

        def _func(*args, **kargs):

            prof = hotshot.Profile("profiling.data")

            res = prof.runcall(func, *args, **kargs)

            prof.close()

            stats = hotshot.stats.load("profiling.data")

            stats.strip_dirs()

            stats.sort_stats('time', 'calls')

            print ">>>---- Begin profiling print"

            stats.print_stats(printlines)

            print ">>>---- End profiling print"

            return res

        return _func

    return _my

Usage:



@profileit(20)

def mop():

    a = 0

    for i in range(100):

        a += mip()

    return a

print mop()