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)

ORM for pys60

korakot (Korakot Chaovavanich) — Fri, 10 Mar 2006 01:30:46 GMT

I have asked for this a few months.
Here's my own first version for an ORM.
It's adapted from SQLAlchemy and SQLObject.
(and whatever library I have searched)

You need to create a database manually first.
My previous snippets (shell, SQL) may help.



from __future__ import generators

import e32db, re

db = e32db.Dbms()

dbv = e32db.Db_view()

db.open(u'C:\\test.db')



# Some helping classes (need more in next version)

class String:

    pass



class Integer:

    pass



class Float:

    pass



class column:

    def __init__(self, coltype):

        self.coltype = coltype

Here's the mother of all your classes that map to database tables.



class Mapper(object):

    def __init__(self, id=None, **kw):

        if id is None:

            self.id = self._insert(**kw)

        else:

            self.id = id

    def _insert(self, **kw):

        names = ','.join(kw.keys())

        values = ','.join([self.quote(k,v) for k,v in kw.items()])

        tablename = self.__class__.__name__

        q = u"INSERT INTO %s(%s) VALUES (%s)" % (tablename, names, values)

        db.execute(q)

        # get last insert ID

        dbv.prepare(db, u'SELECT id FROM '+tablename+' ORDER BY id DESC')

        dbv.first_line()

        dbv.get_line()

        return dbv.col(1)

    def __getattr__(self, name):

        if name in self.mapping.__dict__:

            q = 'SELECT '+name+' FROM '+self.__class__.__name__

            q += ' WHERE id='+str(self.id)

            dbv.prepare(db, unicode(q))

            dbv.first_line()

            dbv.get_line()

            return dbv.col(1)

        else:

            return self.__dict__[name]

    def __repr__(self):

        return '<%s id=%d>' % (self.__class__.__name__, self.id)

    def quote(self, name, value):

        if self.mapping.__dict__[name].coltype == String:

            return "'%s'" % value.replace("'", "''")  # encode single quote

        else:

            return str(value)

    def __setattr__(self, name, value):

        if name in self.mapping.__dict__:

            q = 'UPDATE '+self.__class__.__name__+' SET '+name+'='

            q += self.quote(name, value) + " WHERE id=" + str(self.id)

            db.execute(unicode(q))

        else:

            self.__dict__[name] = value

    def set(self, **kw):

        q = "UPDATE "+self.__class__.__name__+" SET "

        for k, v in kw.items():

            q += k+'='+self.quote(k,v)+','

        q = q[:-1]+" WHERE id=%s" % self.id

        db.execute(unicode(q))

    def delete(self):

        q = 'DELETE FROM '+self.__class__.__name__+" WHERE id=" + str(self.id)

        db.execute(unicode(q))

        self.id = None

    def dict(self):

        names = [k for k in self.mapping.__dict__.keys() if not k.startswith('__')]

        q = 'SELECT '+','.join(names)+' FROM '+self.__class__.__name__

        q += ' WHERE id=' + str(self.id)

        dbv.prepare(db, unicode(q))

        dbv.first_line()

        dbv.get_line()

        dct = {'id': self.id}

        for i in range(dbv.col_count()):

            dct[names[i]] = dbv.col(i+1)

        return dct

    def select(cls, where=None, orderby=None):

        q = 'SELECT id FROM '+cls.__name__

        if where:

            q += ' WHERE '+where

        if orderby:

            q += ' ORDER BY '+orderby

        dbv = e32db.Db_view()  # need its own cursor

        dbv.prepare(db, unicode(q))

        dbv.first_line()

        for i in range(dbv.count_line()):

            dbv.get_line()

            yield cls(dbv.col(1))

            dbv.next_line()

    select = classmethod(select)

Here's how you create your class.



class Person(Mapper):

    class mapping:

        # doesn't need id = column(Integer)

        name = column(String)

        age = column(Integer)

DZone Snippets: orm code

Storm ORM by Canonical in WSGI enabled applications

ORM for pys60