SQLAlchemy BaseModel

j'ai faim

basemodel.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta
from sqlalchemy.orm import sessionmaker, scoped_session
from shimehari.configuration import ConfigManager
from formencode import Invalid

config = ConfigManager.getConfig()
url = 'mysql://' + config['MYSQL_USER'] + ':' + config['MYSQL_PWD'] + '@' + config['MYSQL_HOST'] + '/' + config['MYSQL_NAME'] + '?charset=utf8'
engine = create_engine(url, encoding='utf-8', echo=True)
Session = scoped_session(sessionmaker(autoflush=True, bind=engine))


class Model(object):
    id = Column(Integer, primary_key=True)

    errors = {}
    query = None

    u"""
Lorsque vous essayez d'utiliser une session instanciée
Puisque l'ID de session est étrange, les contre-mesures
    """
    @property
    def session(self):
        return self.query.session

    def __init__(self):
        self.schema = None

    def validate(self, attributes):
        try:
            return self.schema.to_python(attributes)
        except Invalid, e:
            self.errors.update(e.error_dict)
            raise

    def publish(self):
        try:
            self.session.add(self)
            self.session.commit()
        except:
            self.session.rollback()
            raise

    def update(self, attributes):
        for k, v in attributes.iteritems():
            setattr(self, k, v)
        self.publish()

    def delete(self):
        try:
            self.session.delete(self)
            self.session.commit()
        except:
            self.sesssion.rollback()
            raise


def session_mapper(cls):
    cls.query = Session.query(cls)


class BaseDeclarativeMeta(DeclarativeMeta):
    def __init__(cls, classname, bases, dict_):
        DeclarativeMeta.__init__(cls, classname, bases, dict_)
        if not Model in bases:
            session_mapper(cls)

BaseModel = declarative_base(engine, metaclass=BaseDeclarativeMeta, cls=Model)