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)

Storm ORM by Canonical in WSGI enabled applications

sevkin (Vsevolod Balashov) — Fri, 10 Aug 2007 11:39:51 GMT



#!/usr/bin/env python

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

# (c) 2007 Vsevolod Balashov  under terms of LGPL 2.1

# use Storm ORM  in WSGI enabled applications

#

# actual version on http://pypi.python.org/pypi/middlestorm



"""WSGI middleware for Storm.



This is the database access inteface for WSGI enabled (PEP 333) web applications.



Pylons framework example:



in wsgiapp.py



from storm.database import create_database

from middlestorm import MiddleStorm

...



# CUSTOM MIDDLEWARE HERE

app = MiddleStorm(app, create_database(config['app_conf']['sqlalchemy.default.uri']))



in controller:



class DemoController(BaseController):

    def index(self):

        store = request.environ['storm.store']

        ...

"""



from storm.database import Database

from storm.store import Store

from threading import local



__all__ = ["MiddleStorm"]

__author__ = "Vsevolod Balashov "

__version__ = "0.1"



class SingleConn(Database):

    """Database proxy class.

    Share single database connection between all Store object instances in threads.

    If you have readonly database or not use transactions why not?

    """

    def __init__(self, database):

        self._database = database

        self._connection = database.connect()

    def connect(self):

        return self._connection



class MiddleStorm(object):

    """WSGI middleware.

    Add Store object instance in environ['storm.store']. Each thread contains own instance.

    """



    def __init__(self, app, database, single = False):

        """Create WSGI middleware.

        @param app: up level application or middleware.

        @param database: instance of Database returned create_database.

        @param: single: use single database connection in all threads. 

        """

        assert isinstance(database, Database), \

            'database must be subclass of storm.database.Database'

        if single:

            self._database = SingleConn(database)

        else:

            self._database = database

        self._app = app

        self._local = local()



    def __call__(self, environ, start_response):

        try:

            environ['storm.store']  = self._local.store

        except AttributeError:

            environ['storm.store']  = \

                self._local.__dict__.setdefault('store', Store(self._database))

        return self._app(environ, start_response)

DZone Snippets: wsgi code

Pattern Matching based WSGI-enabled URL routing tool.

Storm ORM by Canonical in WSGI enabled applications