How to create and associate another model during creation in flask-sqlalchemy?

Question

I have a one-to-many relationship and I'd like to create and associate the related model to the model being created at creation time.

Here I have User and Item models. When a user is first created, the user should get a default item automatically:

class User(Model):
    id = Column(Integer, primary_key=True)
    name = Column(String, default="", unique=True)
    items = relationship('Item', backref='user', lazy='dynamic')

    def __init__(self, **kwargs):
        super(User, self).__init__(**kwargs)
        self.items.append(...)    # <---- Add default item here?

class Item(Model):
    id = Column(Integer, primary_key=True)
    name = Column(String, default="")
    userid = Column(Integer,  ForeignKey('user.id'))

But I'm not sure how to append the new Item in User.__init__()? Is it something like:

# in User __init__()
item = Item(name="default item")
self.items.append(item)
session.add(item)

But then I"m unclear what happens to the transaction if something fails on either model, and whether I need to call session.add(item) or if appending the item to user automatically does this.

ie, what happens to the transaction when I create a user and this throws an error because name is not unique. Will the item still be created?

def doit():
    myuser = User(name='bob')   # <---- Say "bob" is already in DB
    session.add(myuser)
    session.commit()

What is the valid way of creating a default Item for User at user creation time?

Answer 1

The best way to solve these types of problems is to create some test code. I find prototyping in a sqlite memory database works very well.

So if the relationship is configured correctly - like in your case - sqlalchemy will figure out the key relation and automatically set the foreign key.

But seeing is believing, I have added some sample code that has unittests for the scenarios you describe. Before each test I drop and recreate all tables to ensure the schema is clean.

import logging
logging.basicConfig()

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()


from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String, default="", unique=True)
    items = relationship('Item', backref='user', lazy='dynamic')

    def __init__(self, **kwargs):
        super(User, self).__init__(**kwargs)
        self.items.append(Item(name='DefaultItem'))


class Item(Base):
    __tablename__ = 'item'
    id = Column(Integer, primary_key=True)
    name = Column(String, default="", unique=True)  # Unique so we can create a failing test
    user_id = Column(Integer,  ForeignKey('user.id'))


from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker
from contextlib import contextmanager


engine = create_engine('sqlite://', echo=True)
Session = sessionmaker(bind=engine)


@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


import unittest


class Test(unittest.TestCase):
    def setUp(self):
        Base.metadata.create_all(engine)  # Reset

    def tearDown(self):
        Base.metadata.drop_all(engine)  # Reset

    def test_sanity(self):
        with session_scope() as session:
            self.assertEqual(session.query(User).all(), [])
            self.assertEqual(session.query(Item).all(), [])

    def test_item_automatically_added(self):
        with session_scope() as session:
            user = User(name='John')
            session.add(user)

        with session_scope() as session:
            user = session.query(User).filter(User.name == 'John').one()
            self.assertIsNotNone(user.items.filter(Item.name == 'DefaultItem').one())

    def test_duplicate_item_causes_user_save_to_fail(self):
        with session_scope() as session:
            item = Item(name='DefaultItem')
            session.add(item)

        from sqlalchemy.exc import IntegrityError

        with self.assertRaises(IntegrityError):
            with session_scope() as session:
                user = User(name='John')
                session.add(user)


if __name__ == '__main__':
    unittest.main()

So these tests cover all the scenarios except for one. If the user fails to be created the test will fail. I couldn't think of an easy way to do this test but I did check the sql echo output

INFO:sqlalchemy.engine.base.Engine:INSERT INTO user (name) VALUES (?)
2015-09-15 20:55:49,834 INFO sqlalchemy.engine.base.Engine INSERT INTO user (name) VALUES (?)
INFO:sqlalchemy.engine.base.Engine:('John',)
INFO:sqlalchemy.engine.base.Engine:INSERT INTO item (name, user_id) VALUES (?, ?)
2015-09-15 20:55:49,834 INFO sqlalchemy.engine.base.Engine ('John',)
INFO:sqlalchemy.engine.base.Engine:('DefaultItem', 1)
2015-09-15 20:55:49,834 INFO sqlalchemy.engine.base.Engine INSERT INTO item (name, user_id) VALUES (?, ?)

User is always inserted before item so we cannot get dangling items anyway. To be even more sure of this you could make the item.user_id foreign key not nullable.