DZone Snippets: Sevkin's Code Snippets

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

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)

Ruby port of range2cidr. Convert IP ranges to set of CIDR.

sevkin (Vsevolod Balashov) — Wed, 11 Apr 2007 19:32:53 GMT

See rfc1878 for more detail about CIDR



#!/usr/bin/env ruby

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

# $Hg: range_cidr.rb,v af9566d89389 2007-04-12 20:28 +0400 $

# (C) 2007 under terms of LGPL v2.1

# by Vsevolod S. Balashov 

#

# backported from perl code

# http://www.irbs.net/internet/postfix/0401/att-3032/cidr_range.pl.gz



def range_cidr(first, last, &block)

  if first < last

    idx1 = 32

    idx1 -= 1 while first[idx1] == last[idx1]

    prefix = first >> idx1+1 << idx1+1



    idx2 = 0

    idx2 += 1 while idx2 <= idx1 and first[idx2] == 0 and last[idx2] == 1



    if idx2 <= idx1

      range_cidr(first, prefix | 2**idx1-1, &block)

      range_cidr(prefix | 1 << idx1, last, &block)

    else

      yield prefix, 32-idx2

    end

  else

    yield first, 32

  end

end

Example Usage



#!/usr/bin/env ruby

# $Hg: range2cidr.rb,v 2142c33ada8b 2007-04-11 23:14 +0400 $

# (C) 2007 under terms of GPL v2

# by Vsevolod S. Balashov 

#

# example usage of range_cidr.rb



require 'lib/range_cidr'

require 'ipaddr'

require 'socket'



if __FILE__ == $0

  if ARGV.size == 2

    range_cidr(IPAddr.new(ARGV[0]).to_i, IPAddr.new(ARGV[1]).to_i) { |subnet, mask|

      puts "#{IPAddr.new(subnet, Socket::AF_INET).to_s}/#{mask}"  }

  else

    puts "usage: range2cidr  "

    puts "example: range2cidr 192.168.1.0 192.168.2.255"

  end

end

Type in terminal and look result



$ ruby range2cidr.rb 192.168.1.0 192.168.2.255

192.168.1.0/24

192.168.2.0/24

I hate perl.

Google PageRank Ruby Checker

sevkin (Vsevolod Balashov) — Mon, 15 Jan 2007 15:59:31 GMT



#!/usr/bin/env ruby

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

# $Id: google-pr.rb,v b205c14e4ef5 2007-01-15 13:53 +0300 $

# (C) 2006-2007 under terms of LGPL v2.1 

# by Vsevolod S. Balashov 

# based on 3rd party code snippets (see comments)



require 'uri'

require 'open-uri'



module SEO



  # http://blog.outer-court.com/archive/2004_06_27_index.html#108834386239051706

  class GooglePR



    def initialize(uri)

      @uri = uri

    end



    M=0x100000000 # modulo for unsigned int 32bit(4byte)



    def m1(a,b,c,d)

      (((a+(M-b)+(M-c))%M)^(d%M))%M # mix/power mod

    end



    def i2c(i)

      [i&0xff, i>>8&0xff, i>>16&0xff, i>>24&0xff]

    end



    def c2i(s,k=0)

      ((s[k+3].to_i*0x100+s[k+2].to_i)*0x100+s[k+1].to_i)*0x100+s[k].to_i

    end



    def mix(a,b,c)

      a = a%M; b = b%M; c = c%M

      a = m1(a,b,c, c >> 13); b = m1(b,c,a, a <<  8); c = m1(c,a,b, b >> 13)

      a = m1(a,b,c, c >> 12); b = m1(b,c,a, a << 16); c = m1(c,a,b, b >>  5)

      a = m1(a,b,c, c >>  3); b = m1(b,c,a, a << 10); c = m1(c,a,b, b >> 15)

      [a, b, c]

    end



    def old_cn(iurl = 'info:' + @uri)

      a = 0x9E3779B9; b = 0x9E3779B9; c = 0xE6359A60

      len = iurl.size 

      k = 0

      while (len >= k + 12) do

        a += c2i(iurl,k); b += c2i(iurl,k+4); c += c2i(iurl,k+8)

        a, b, c = mix(a, b, c)

        k = k + 12

      end

      a += c2i(iurl,k); b += c2i(iurl,k+4); c += (c2i(iurl,k+8) << 8) + len

      a,b,c = mix(a,b,c)

      return c

    end



    def cn

      ch = old_cn

      ch = ((ch/7) << 2) | ((ch-(ch/13).floor*13)&7)

      new_url = []

      20.times { i2c(ch).each { |i| new_url << i }; ch -= 9 }

      ('6' + old_cn(new_url).to_s).to_i

    end



    def request_uri

      # http://www.bigbold.com/snippets/posts/show/1260 + _ -> %5F

      "http://toolbarqueries.google.com/search?client=navclient-auto&hl=en&ch=#{cn}&ie=UTF-8&oe=UTF-8&features=Rank&q=info:#{URI.escape(@uri, /[^-.!~*'()a-zA-Z\d]/)}"

    end

 

    def page_rank(uri = @uri)

      @uri = uri if uri != @uri

      open(request_uri) { |f| return $1.to_i if f.string =~ /Rank_1:\d:(\d+)/ }

      nil

    end



    private :m1, :i2c, :c2i, :mix, :old_cn

    attr_accessor :uri

  end



end



if __FILE__ == $0 and 1 == ARGV.size

  puts SEO::GooglePR.new(ARGV[0]).page_rank

end

nowww!

sevkin (Vsevolod Balashov) — Thu, 21 Dec 2006 17:57:57 GMT

set redirect from www.example.com to example.com for nginx web server



server {

    listen       80;

    server_name  example.com www.example.com;

    if ( $host = www.example.com ) {

        rewrite ^\/(.*)$ http://example.com/$1 permanent;

    }



    .....

}

backup subversion repository over email

sevkin (Vsevolod Balashov) — Thu, 21 Sep 2006 22:58:18 GMT



#!/bin/bash

#

# $Id: repodiff 3 2006-09-21 18:48:39Z sevkin $

#

# subversion repository incremental backup over e-mail

#

# (c) 2006 Vsevolod Balashov under terms of GNU GPL v.2 or later



SVNROOT=/var/svn

EMAIL=your@email.here

STORE=`mktemp -d`

GPGCRYPT=n



for REPO in `ls $SVNROOT`; do 

	REPOPATH=$SVNROOT/$REPO;

	if [ -r $REPOPATH/youngest ]; then

		LATEST=`cat $REPOPATH/youngest`

		YOUNGEST=`svnlook youngest $REPOPATH`

		if [ $LATEST -lt $YOUNGEST ]; then

			svnadmin dump $REPOPATH --incremental -r $LATEST:$YOUNGEST >$STORE/$REPO 2>/dev/null

		fi

	else

		svnadmin dump $REPOPATH --incremental >$STORE/$REPO 2>/dev/null

	fi 

	echo $YOUNGEST >$REPOPATH/youngest 

done



if [ `ls $STORE | wc -w` -gt 0 ]; then

	BACKUP=repodiff_`date -u +%Y%m%d%H%M%S`.tar.bz2

	ATTACH=$STORE/../$BACKUP

	tar  -C $STORE -cjf $ATTACH .

	if [ $GPGCRYPT = y ]; then

		gpg -e -r $EMAIL $ATTACH

		ATTACH=$ATTACH.gpg

	fi

	echo "." | mutt -c $EMAIL -a $ATTACH -s "repository incremental backup"

	rm -f $ATTACH

fi



rm -rf $STORE

Subversion service for low-load (personal?) sources repository

sevkin (Vsevolod Balashov) — Tue, 19 Sep 2006 23:48:50 GMT

Create separate user and insert svnserve into inetd.



# useradd -g root -s /bin/false -d /dev/null -c "SubVersion Daemon" svnserve

# mkdir /var/svn

# chown -R svnserve /var/svn

# update-inetd --add 'svn\tstream\ttcp\tnowait\tsvnserve\t/usr/sbin/tcpd\t/usr/bin/svnserve --inetd --root /var/svn'

/var/svn - root of repository.
update-inetd - standart tool in debian and ubuntu linux distros

You must run svnadmin as svnserve user for manage your repository



$ sudo sudo -u svnserve svnadmin