rendered paste bodyfrom sqlalchemy import *
from sqlalchemy.orm import *
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
tablename = cls.__tablename__
our_stuff = {}
for k in dict_:
value = dict_[k]
if not isinstance(value, (Column, MapperProperty)):
continue
prop = cls.__deferred_relation(value)
our_stuff[k] = prop
cls.__table__ = Table(tablename, cls.metadata, *[
c for c in our_stuff.values() if isinstance(c, Column)
])
cls.mapper = mapper(cls, cls.__table__, properties=our_stuff)
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 isinstance(value, Column) and hasattr(cls, 'mapper'):
raise exceptions.InvalidRequestError("Can't configure new Columns on an already-completed class")
elif isinstance(value, MapperProperty) and hasattr(cls, 'mapper'):
cls.mapper.add_property(key, cls.__deferred_relation(value))
else:
type.__setattr__(cls, key, value)
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__':
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")
class Address(Base):
__tablename__ = 'addresses'
id = Column('id', Integer, primary_key=True)
email = Column('email', String(50))
user_id = Column('user_id', Integer, ForeignKey('users.id'))
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'))
)
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)
sess.flush()
print sess.query(User).options(eagerload('addresses')).all()