Never been to DZone Snippets before?

Snippets is a public source code repository. Easily build up your personal collection of code snippets, categorize them with tags / keywords, and share them with the world

About this user

Vsevolod Balashov

« Newer Snippets
Older Snippets »
Showing 1-2 of 2 total  RSS 

Pattern Matching based WSGI-enabled URL routing tool.

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/<id:\d+>')
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 <vsevolod@balashov.name>
# 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

#!/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 <vsevolod@balashov.name> 

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
« Newer Snippets
Older Snippets »
Showing 1-2 of 2 total  RSS