rendered paste body
from sqlalchemy.schema import Table, SchemaItem, Column
from sqlalchemy.orm import synonym as _orm_synonym
from sqlalchemy.orm.interfaces import MapperProperty
from sqlalchemy.orm.properties import PropertyLoader
from sqlalchemy import exceptions
class Thingy(type):
def __init__(cls, classname, bases, dict_):
if '_thingy_class_registry' in cls.__dict__:
return type.__init__(cls, classname, bases, dict_)
cls._thingy_class_registry[classname] = cls
our_stuff = {}
for k in dict_:
value = dict_[k]
if not isinstance(value, (Column, MapperProperty, declared_synonym)):
continue
if isinstance(value, declared_synonym):
value._setup(cls, k, our_stuff)
else:
prop = cls.__deferred_relation(value)
our_stuff[k] = prop
table = None
if '__table__' not in cls.__dict__:
if '__tablename__' in cls.__dict__:
tablename = cls.__tablename__
cls.__table__ = table = Table(tablename, cls.metadata, *[
c for c in our_stuff.values() if isinstance(c, Column)
])
else:
table = cls.__table__
inherits = cls.__mro__[1]
inherits = cls._thingy_class_registry.get(inherits.__name__, None)
mapper_args = getattr(cls, '__mapper_args__', {})
cls.mapper = mapper(cls, cls.__table__, inherits=inherits, properties=our_stuff, **mapper_args)
return type.__init__(cls, classname, bases, dict_)
def __deferred_relation(cls, prop):
if isinstance(prop, PropertyLoader) and isinstance(prop.argument, basestring):
arg = prop.argument
def return_cls():
return cls._thingy_class_registry[arg]
prop.argument = return_cls
return prop
def __setattr__(cls, key, value):
if hasattr(cls, 'mapper'):
if isinstance(value, Column):
cls.__table__.append_column(value)
cls.mapper.add_property(key, value)
elif isinstance(value, MapperProperty):
cls.mapper.add_property(key, cls.__deferred_relation(value))
elif isinstance(value, declared_synonym):
value._setup(cls, key, None)
else:
type.__setattr__(cls, key, value)
else:
type.__setattr__(cls, key, value)
class declared_synonym(object):
def __init__(self, prop, name, mapperprop):
self.prop = prop
self.name = name
self.mapperprop = mapperprop
def _setup(self, cls, key, init_dict):
prop = self.mapperprop
prop = cls._Thingy__deferred_relation(prop)
if init_dict:
init_dict[self.name] = prop
init_dict[key] = _orm_synonym(self.name)
else:
setattr(cls, self.name, prop)
setattr(cls, key, _orm_synonym(self.name))
setattr(cls, key, self.prop)
def declarative_base(engine=None, metadata=None):
lcl_metadata = metadata or MetaData()
class Base(object):
__metaclass__ = Thingy
metadata = lcl_metadata
if engine:
metadata.bind = engine
_thingy_class_registry = {}
def __init__(self, **kwargs):
for k in kwargs:
setattr(self, k, kwargs[k])
return Base
if __name__ == '__main__':
from sqlalchemy import *
from sqlalchemy.orm import *
Base = declarative_base(create_engine('sqlite://', echo=True))
class User(Base):
__tablename__ = 'users'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
addresses = relation("Address", backref="user")
type = Column('type', String(50))
__mapper_args__ = dict(polymorphic_on=type, polymorphic_identity='user')
class AdminUser(User):
__tablename__ = 'admin_users'
__mapper_args__ = dict(polymorphic_identity='adminuser')
id = Column('id', Integer, ForeignKey('users.id'), primary_key=True)
supername = Column('supername', String(50))
class Address(Base):
__tablename__ = 'addresses'
id = Column('id', Integer, primary_key=True)
user_id = Column('user_id', Integer, ForeignKey('users.id'))
def _set_email(self, email):
self._email = email
def _get_email(self):
return self._email
email = property(_get_email, _set_email)
email = declared_synonym(email, '_email', Column('email', String(50)))
class Keyword(Base):
__tablename__ = 'keywords'
id = Column('id', Integer, primary_key=True)
name = Column('name', String(50))
user_keywords = Table('user_keywords', Base.metadata,
Column('user_id', Integer, ForeignKey('users.id')),
Column('keyword_id', Integer, ForeignKey('keywords.id'))
)
# test adding relations after the fact
User.keywords = relation(Keyword, secondary=user_keywords, backref='users')
Base.metadata.create_all()
sess = create_session()
u1 = User(name='ed',
addresses=[Address(email='eds email')],
keywords = [Keyword(name='one'), Keyword(name='two')]
)
sess.save(u1)
a1 = AdminUser(name='some admin', supername='root')
sess.save(a1)
sess.flush()
print sess.query(User).filter(User.name=='ed').all()
print sess.query(User).filter(User.addresses.any(Address.email.like('%ed%'))).all()
print sess.query(User).with_polymorphic('*').all()