# -*- coding: utf-8 -*- "" # Copyright (C) 2014 Team tiramisu (see AUTHORS for all contributors) # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # # ____________________________________________________________ from tiramisu.i18n import _ from tiramisu.setting import groups from sqlalchemy import not_, or_ from sqlalchemy.ext.declarative import declarative_base, declared_attr from sqlalchemy.ext.associationproxy import association_proxy from sqlalchemy import create_engine, Column, Integer, String, Boolean, \ PickleType, ForeignKey, Table from sqlalchemy.orm import relationship, backref from sqlalchemy.orm import sessionmaker from sqlalchemy.orm.collections import attribute_mapped_collection #FIXME engine = create_engine('sqlite:///:memory:') SqlAlchemyBase = declarative_base() #FIXME a voir: # # Organization.members will be a Query object - no loading # # of the entire collection occurs unless requested # lazy="dynamic", #____________________________________________________________ # # require def load_requires(collection_type, proxy): def getter(obj): if obj is None: return None ret = [] requires = getattr(obj, proxy.value_attr) for require in requires: option = session.query(_Base).filter_by(id=require.option).first() ret.append(tuple([option, require.expected, require.action, require.inverse, require.transitive, require.same_action])) return tuple(ret) def setter(obj, value): setattr(obj, proxy.value_attr, value) return getter, setter class _Require(SqlAlchemyBase): __tablename__ = "require" id = Column(Integer, primary_key=True) requires_id = Column(Integer, ForeignKey("baseoption.id"), nullable=False) requires = relationship('_RequireOption') def __init__(self, requires): for require in requires: self.requires.append(_RequireOption(require)) class _RequireOption(SqlAlchemyBase): __tablename__ = 'requireoption' id = Column(Integer, primary_key=True) require_id = Column(Integer, ForeignKey("require.id"), nullable=False) option = Column(Integer, nullable=False) _expected = relationship("_RequireExpected", collection_class=list, cascade="all, delete-orphan") expected = association_proxy("_expected", "expected") #expected = Column(String) action = Column(String, nullable=False) inverse = Column(Boolean, default=False) transitive = Column(Boolean, default=True) same_action = Column(Boolean, default=True) def __init__(self, values): option, expected, action, inverse, transitive, same_action = values self.option = option.id self.expected = expected self.action = action self.inverse = inverse self.transitive = transitive self.same_action = same_action class _RequireExpected(SqlAlchemyBase): __tablename__ = 'expected' id = Column(Integer, primary_key=True) require = Column(Integer, ForeignKey('requireoption.id'), nullable=False) expected = Column(PickleType) def __init__(self, expected): #FIXME ne pas creer plusieurs fois la meme _expected_ #FIXME pareil avec calc_properties self.expected = expected class _CalcProperties(SqlAlchemyBase): __tablename__ = 'calcproperty' id = Column(Integer, primary_key=True) require = Column(Integer, ForeignKey('baseoption.id'), nullable=False) name = Column(PickleType) def __init__(self, name): #FIXME ne pas creer plusieurs fois la meme _expected_ #FIXME pareil avec calc_properties self.name = name #____________________________________________________________ # # properties class _PropertyOption(SqlAlchemyBase): __tablename__ = 'propertyoption' id = Column(Integer, primary_key=True) option = Column(Integer, ForeignKey('baseoption.id'), nullable=False) name = Column(String) def __init__(self, name): self.name = name #____________________________________________________________ # # information class _Information(SqlAlchemyBase): __tablename__ = 'information' id = Column(Integer, primary_key=True) option = Column(Integer, ForeignKey('baseoption.id'), nullable=False) key = Column(String) value = Column(PickleType) def __init__(self, key, value): self.key = key self.value = value #____________________________________________________________ # # callback def load_callback_parm(collection_type, proxy): def getter(obj): if obj is None: return None ret = [] requires = getattr(obj, proxy.value_attr) for require in requires: if require.value is not None: ret.append(require.value) else: option = session.query(_Base).filter_by(id=require.option).first() ret.append((option, require.force_permissive)) return tuple(ret) def setter(obj, value): setattr(obj, proxy.value_attr, value) return getter, setter class _CallbackParamOption(SqlAlchemyBase): __tablename__ = 'callback_param_option' id = Column(Integer, primary_key=True) callback_param = Column(Integer, ForeignKey('callback_param.id')) option = Column(Integer) force_permissive = Column(Boolean) value = Column(PickleType) def __init__(self, option=None, force_permissive=None, value=None): if value is not None: self.value = value else: self.option = option.id self.force_permissive = force_permissive class _CallbackParam(SqlAlchemyBase): __tablename__ = 'callback_param' id = Column(Integer, primary_key=True) callback = Column(Integer, ForeignKey('baseoption.id')) key = Column(String) params = relationship('_CallbackParamOption') def __init__(self, key, params): self.key = key for param in params: if isinstance(param, tuple): self.params.append(_CallbackParamOption(option=param[0], force_permissive=param[1])) else: self.params.append(_CallbackParamOption(value=param)) #____________________________________________________________ # # consistency consistency_table = Table('consistencyopt', SqlAlchemyBase.metadata, Column('left_id', Integer, ForeignKey('consistency.id')), Column('right_id', Integer, ForeignKey('baseoption.id')) ) class _Consistency(SqlAlchemyBase): __tablename__ = 'consistency' id = Column(Integer, primary_key=True) func = Column(PickleType) params = Column(PickleType) def __init__(self, func, all_cons_opts): self.func = func for option in all_cons_opts: option._consistencies.append(self) self.params = params class _Parent(SqlAlchemyBase): __tablename__ = 'parent' id = Column(Integer, primary_key=True) child_id = Column(Integer) child_name = Column(String) parent_id = Column(Integer) def __init__(self, parent, child): self.parent_id = parent.id self.child_id = child.id self.child_name = child._name #____________________________________________________________ # # Base class _Base(SqlAlchemyBase): __tablename__ = 'baseoption' id = Column(Integer, primary_key=True) _name = Column(String) #FIXME not autoload _infos = relationship("_Information", collection_class=attribute_mapped_collection('key'), cascade="all, delete-orphan") _informations = association_proxy("_infos", "value") _default = Column(PickleType) _default_multi = Column(PickleType) _reqs = relationship("_Require", collection_class=list) _requires = association_proxy("_reqs", "requires", getset_factory=load_requires) _multi = Column(Boolean) _multitype = Column(String) ###### _callback = Column(PickleType) _call_params = relationship('_CallbackParam', collection_class= attribute_mapped_collection('key')) _callback_params = association_proxy("_call_params", "params", getset_factory=load_callback_parm) _validator = Column(PickleType) _val_params = relationship('_CallbackParam', collection_class= attribute_mapped_collection('key')) _validator_params = association_proxy("_call_params", "params", getset_factory=load_callback_parm) ###### #FIXME pas 2 fois la meme properties dans la base ... #FIXME not autoload #FIXME normalement tuple ... transforme en set ! _props = relationship("_PropertyOption", collection_class=set) _properties = association_proxy("_props", "name") #FIXME fusion avec expected _calc_props = relationship("_CalcProperties", collection_class=set) _calc_properties = association_proxy("_calc_props", "name") _warnings_only = Column(Boolean) _readonly = Column(Boolean, default=False) _consistencies = relationship('_Consistency', secondary=consistency_table, backref=backref('options', enable_typechecks=False)) _type = Column(String(50)) __mapper_args__ = { 'polymorphic_identity': 'option', 'polymorphic_on': _type } _extra = Column(PickleType) #FIXME devrait etre une table _group_type = Column(String) _is_build_cache = Column(Boolean, default=False) def __init__(self): self.commit() def commit(self): session.add(self) session.commit() def _add_consistency(self, func, all_cons_opts, params): _Consistency(func, all_cons_opts, params) def _get_consistencies(self): return [(consistency.func, consistency.options, consistency.params) for consistency in self._consistencies] def _get_id(self): return self.id # ____________________________________________________________ # information #def impl_set_information(self, key, value): # """updates the information's attribute # (which is a dictionary) # :param key: information's key (ex: "help", "doc" # :param value: information's value (ex: "the help string") # """ # info = session.query(_Information).filter_by(option=self.id, key=key).first() # #FIXME pas append ! remplacer ! # if info is None: # self._informations.append(_Information(key, value)) # else: # info.value = value #def impl_get_information(self, key, default=None): # """retrieves one information's item # :param key: the item string (ex: "help") # """ # info = session.query(_Information).filter_by(option=self.id, key=key).first() # if info is not None: # return info.value # return self._informations[key] # elif default is not None: # return default # else: # raise ValueError(_("information's item not found: {0}").format( # key)) class Cache(SqlAlchemyBase): __tablename__ = 'cache' id = Column(Integer, primary_key=True) #FIXME indexer ... les 3 path = Column(String, nullable=False) descr = Column(Integer, nullable=False) option = Column(Integer, nullable=False) opt_type = Column(String, nullable=False) def __init__(self, descr, option, path): self.descr = descr.id self.option = option.id self.path = path self.opt_type = option.__class__.__name__ class StorageOptionDescription(object): def impl_already_build_caches(self): return self._is_build_cache def impl_get_opt_by_path(self, path): ret = session.query(Cache).filter_by(descr=self.id, path=path).first() if ret is None: raise AttributeError(_('no option for path {0}').format(path)) return session.query(_Base).filter_by(id=ret.option).first() def impl_get_path_by_opt(self, opt): ret = session.query(Cache).filter_by(descr=self.id, option=opt.id).first() if ret is None: raise AttributeError(_('no option {0} found').format(opt)) return ret.path def impl_get_group_type(self): return getattr(groups, self._group_type) def impl_build_cache_option(self, descr=None, _currpath=None): if descr is None: save = True descr = self _currpath = [] else: save = False for option in self.impl_getchildren(): attr = option.impl_getname() session.add(Cache(descr, option, str('.'.join(_currpath + [attr])))) if isinstance(option, StorageOptionDescription): _currpath.append(attr) option.impl_build_cache_option(descr, _currpath) _currpath.pop() if save: self._is_build_cache = True session.commit() def impl_get_options_paths(self, bytype, byname, _subpath, only_first): sqlquery = session.query(Cache).filter_by(descr=self.id) if bytype is None: sqlquery = sqlquery.filter(not_(Cache.opt_type == 'OptionDescription')) else: sqlquery = sqlquery.filter_by(opt_type=bytype.__name__) query = '' or_query = '' if _subpath is not None: query += _subpath + '.' if byname is not None: or_query = query + byname query += '%.' + byname if query != '': filter_query = Cache.path.like(query) if or_query != '': filter_query = or_(Cache.path == or_query, filter_query) sqlquery = sqlquery.filter(filter_query) if only_first: opt = sqlquery.first() if opt is None: return tuple() option = session.query(_Base).filter_by(id=opt.option).first() return ((opt.path, option),) else: ret = [] for opt in sqlquery.all(): option = session.query(_Base).filter_by(id=opt.option).first() ret.append((opt.path, option)) return ret def _add_children(self, child_names, children): for child in children: session.add(_Parent(self, child)) def impl_getchildren(self): for child in session.query(_Parent).filter_by(parent_id=self.id).all(): yield(session.query(_Base).filter_by(id=child.child_id).first()) #return def __getattr__(self, name): if name.startswith('_') or name.startswith('impl_'): return object.__getattribute__(self, name) child = session.query(_Parent).filter_by(parent_id=self.id, child_name=name).first() if child is None: raise AttributeError(_('unknown Option {0} ' 'in OptionDescription {1}' '').format(name, self.impl_getname())) return session.query(_Base).filter_by(id=child.child_id).first() class StorageBase(_Base): @declared_attr def __mapper_args__(self): return {'polymorphic_identity': self.__name__.lower()} #engine.echo = True SqlAlchemyBase.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session()