SchemAlchemy https://github.com/schematics/schemalchemy/issues/1

test-many.py

from schemalchemy import Base
from schematics.types import StringType
from schematics.types.compound import ListType, ModelType
from schematics.types.serializable import serializable

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, relationship, synonym

DATABASE = 'test.db'


class ChildRecord(Base):
    #SQLAlchemy Table Definition
    __tablename__ = 'child'

    _child_id = Column('child_id', Integer, primary_key=True)
    _parent_id = Column('parent_id', String, ForeignKey('parent.id'))
    _type = Column('type', String)
    _value = Column('value', String)

    parent_id = synonym('_parent_id')

    #Schematics Definition
    type = StringType()
    value = StringType()


class ParentRecord(Base):
    #SQLAlchemy Table Definition
    __tablename__ = 'parent'

    _id = Column('id', String, primary_key=True, nullable=False)
    child = relationship('ChildRecord', collection_class=list,
        cascade='all, delete-orphan', passive_deletes=True)

    #Schematics Definition
    id = StringType(default='1')

    @serializable(type=ListType(ModelType(ChildRecord)))
    def children(self):
        return self.child


def fixture_child():
    p = ChildRecord({'type': 'Child1', 'value': 'child data'})
    return p


def fixture_parent():
    c = ParentRecord()
    c.id = '2'
    c.validate()
    return c


def init_db():
    engine = create_engine('sqlite:///' + DATABASE)
    session = sessionmaker(bind=engine)
    Base.metadata.create_all(engine)
    return session


def test_insert_data(session):
    p = fixture_parent()
    c = fixture_child()
    session.add(p)
    session.add(c)
    session.commit()
    print 'Created:'
    print '\tParent', p.serialize()
    print '\tChild ', c.serialize()


def test_load_data(session):
    p = session.query(ParentRecord).first()
    c = session.query(ChildRecord).first()
    fp = fixture_parent()
    fc = fixture_child()
    assert_equal(p.serialize(), fp.serialize())
    assert_equal(c.serialize(), fc.serialize())
    print 'Loaded:'
    print '\tParent', p.serialize()
    print '\tChild', c.serialize()


def test_update_data(Session):
    sessionChild = Session()
    c = sessionChild.query(ChildRecord).first()
    sessionParent = Session()
    p = sessionParent.query(ParentRecord).first()
    pc = sessionParent.merge(c)
    p.child.append(pc)
    sessionParent.commit()
    session = Session()
    p = session.query(ParentRecord).first()
    c = session.query(ChildRecord).first()
    fp = fixture_parent()
    fc = fixture_child()
    fp.child.append(fc)
    assert_equal(p.serialize(), fp.serialize())
    assert_equal(c.serialize(), fc.serialize())
    print 'Updated:'
    print '\tParent ', p.serialize()
    print '\tChild', c.serialize()


def assert_equal(v1, v2):
    assert v1 == v2, '%s != %s' % (v1, v2)


if __name__ == '__main__':
    import os
    if os.path.exists(DATABASE):
        os.remove(DATABASE)
    Session = init_db()
    test_insert_data(Session())
    test_load_data(Session())
    test_update_data(Session)
    print 'All tests OK'