How do I setup dependent factories using Factory Boy and Flask-SQLAlchemy?

Question

What is the correct way to use factory boy with Flask-SQLAlchemy and Foreign Key constraints?

Consider the following Flask SQLAlchemy Model setup:

# coding=utf-8
from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)

# ------------------------------
#   SQLAlchemy Table Models
# ------------------------------
class User(db.Model):
    """ A SQLAlchemy simple model class who represents a user with a ForeignKey Constraint"""
    __tablename__ = 'UserTable'

    user_pk = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.Unicode(20))
    group_fk = db.Column(db.ForeignKey("GroupTable.group_pk"), nullable=False)


class Group(db.Model):
    """ A SQLAlchemy simple model class who represents a user """
    __tablename__ = 'GroupTable'

    group_pk = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.String(), nullable=False)


# -------------------------
#   Create the SQL tables
# -------------------------
db.create_all()

The User schema requires a Group foreign key when creating a new user. Since the Group primary key is assigned from the database, the factory would need to commit a group entry and get the entry's primary key so that it can provide it to the new User.

How do I create a group, save it to the DB, and provide it's key to the User Factory?

Factory Boy has examples for dealing with Foreign Keys but they don't seem to apply to SQLAlchemy. Here are the factories and the failure point:

# ----------------------------------------
#   Factory-Boy User and Group Factories
# ----------------------------------------
from factory import alchemy, Sequence, RelatedFactory


class GroupFactory(alchemy.SQLAlchemyModelFactory):
    class Meta(object):
        model = Group
        sqlalchemy_session = db.session  # the SQLAlchemy session object

    name = Sequence(lambda n: "Group {}".format(n))
    # group_pk = Sequence(lambda n: n)


class UserFactory(alchemy.SQLAlchemyModelFactory):
    class Meta(object):
        model = User
        sqlalchemy_session = db.session  # the SQLAlchemy session object

    user_pk = Sequence(lambda n: n)
    name = Sequence(lambda n: u'User %d' % n)  # coding=utf-8
    group_fk = RelatedFactory(GroupFactory)


# ----------------------
#   Factory tests
# ----------------------
# Create a new Group from our factory
group_from_factory = GroupFactory(name='a new group name')
assert group_from_factory.group_pk is None
# Save it to our DB
db.session.add(group_from_factory)
db.session.commit()

# Verify that Group Saved correctly to DB
group_from_db = db.session.query(Group).filter(Group.group_pk == group_from_factory.group_pk).first()
assert group_from_db.group_pk is not None
assert group_from_db.name == 'a new group name'
assert group_from_db.group_pk == group_from_factory.group_pk

# Create a new User from our factory
user_from_factory = UserFactory()
db.session.add(user_from_factory)
# ----------------------------------------------
#   FAILS AT COMMIT() - NOT NULL constraint failed (group_fk is null)
# ----------------------------------------------
db.session.commit()

assert user_from_factory.user_pk is not None
assert user_from_factory.name is not None
assert user_from_factory.group_fk is not None

Junna

SubFactory doesn't work. See comments on this. According to the Factory Boy api it should, but it doesn't which is why I posted on this. – Junna

Answer 1

The issue comes from using a RelatedFactory: those are intended for reverse ForeignKey relations (e.g if you want to build a Group object which already contains a User).

For a direct ForeignKey - like the relation from User to Group, use a SubFactory:

class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = db.session

    # No need to force the user_pk, it is built automatically from the database
    # user_pk = Sequence(lambda n: n)
    name = Sequence(lambda n: u'User %d' % n)  # coding=utf-8
    group_fk = factory.SubFactory(GroupFactory)

I'm not very familiar with Flask-SQLAlchemy, but I've just added a small example to the repository (at https://github.com/rbarrois/factory_boy/tree/master/examples/flask_alchemy) that works yet is quite similar to your situation.

Lukash

Thanks for the suggestion but SubFactory doesn't solve this. SQL throws an 'unsupported type' error since the SubFactory returns a Group instance instead of an Integer. So what I need is to communicate the Group.group_pk Integer value to the User.group_fk attribute so that SQL doesn't freak out when it executes the INSERT statement. Also the Group object returned from SubFactory is not saved in the database yet – Junna

sqlalchemy.exc.InterfaceError: (sqlite3.InterfaceError) Error binding parameter 1 - probably unsupported type. [SQL: u'INSERT INTO "UserTable" (name, group_fk) VALUES (?, ?)'] [parameters: (u'User 0', <__main__.Group object at 0x101c20650>)] – Junna

The github code you wrote is great. It looks like this may be the way to do it. It requires a change in the SQLAlchemy table model which is why I didn't see SubFactory as a solution. It's a solution if the table model is constructed in a particular fashion, which is a better construction. I need more time to review it before I mark the answer, but this is good. – Junna

Answer 2

Xelnor's git link shows the best answer so far, but it requires changes to the sqlalchemy model. Here is the finished working copy of my post using Xelnor's solution:

# coding=utf-8
from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)

SQLAlchemy Table Models

class User(db.Model):
    """ A SQLAlchemy simple model class who represents a user with a ForeignKey Constraint"""
    __tablename__ = 'user'

    id = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.Unicode(20))

    group_id = db.Column(db.Integer, db.ForeignKey('group.id'), nullable=False)

The 'group' db.relationship is what makes the SubFactory call work. UserFactory passes group to the User model, which is setup with this relationship() definition.

    group = db.relationship('Group', backref=db.backref('groups', lazy='dynamic'))

    def __init__(self, name, group):
        self.name = name
        self.group = group

    def __repr__(self):
        return '<Group for %r: %s>' % (self.group, self.name)


class Group(db.Model):
    """ A SQLAlchemy simple model class who represents a user """
    __tablename__ = 'group'

    id = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.String(), nullable=False)

    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return '<Group %r>' % self.name

Create the SQL tables

db.create_all()

Factory-Boy User and Group Factories

from factory import alchemy, Sequence, SubFactory, fuzzy


class BaseFactory(alchemy.SQLAlchemyModelFactory):
    class Meta(object):
        abstract = True
        sqlalchemy_session = db.session


class GroupFactory(BaseFactory):
    class Meta(object):
        model = Group
        sqlalchemy_session = db.session  # the SQLAlchemy session object

    name = fuzzy.FuzzyText()


class UserFactory(BaseFactory):
    class Meta:
        model = User
        sqlalchemy_session = db.session

    name = fuzzy.FuzzyText()
    group = SubFactory(GroupFactory)

Factory tests

# Create a new Group from our factory
group_from_factory = GroupFactory(name='a new group name')
assert group_from_factory.id is None
# Save it to our DB
db.session.add(group_from_factory)
db.session.commit()

# Verify that Group Saved correctly to DB
group_from_db = db.session.query(Group).filter(Group.id == group_from_factory.id).first()
assert group_from_db.id is not None
assert group_from_db.name == 'a new group name'
assert group_from_db.id == group_from_factory.id

# Create a new User from our factory
user1_from_factory = UserFactory(name=u'first')
user2_from_factory = UserFactory(name=u'second')
db.session.add(user1_from_factory)
db.session.add(user2_from_factory)
db.session.commit()

assert user1_from_factory.id is not None
assert user1_from_factory.name is not None
assert user1_from_factory.id is not None

assert user2_from_factory.id is not None
assert user2_from_factory.name is not None
assert user2_from_factory.id is not None

Junna

Answer 3

You can use LazyAttribute and a lambda to generate a your new Group and then pull it's 'group_pk' out.

Working version of your code below:

# coding=utf-8
from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)

# ------------------------------
#   SQLAlchemy Table Models
# ------------------------------
class User(db.Model):
    """ A SQLAlchemy simple model class who represents a user with a ForeignKey Constraint"""
    __tablename__ = 'UserTable'

    user_pk = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.Unicode(20))
    group_fk = db.Column(db.ForeignKey("GroupTable.group_pk"), nullable=False)


class Group(db.Model):
    """ A SQLAlchemy simple model class who represents a user """
    __tablename__ = 'GroupTable'

    group_pk = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.String(), nullable=False)


# -------------------------
#   Create the SQL tables
# -------------------------
db.drop_all()
db.create_all()
# ----------------------------------------
#   Factory-Boy User and Group Factories
# ----------------------------------------
from factory import alchemy, Sequence, LazyAttribute


class GroupFactory(alchemy.SQLAlchemyModelFactory):
    class Meta(object):
        model = Group
        sqlalchemy_session = db.session  # the SQLAlchemy session object

    name = Sequence(lambda n: "Group {}".format(n))
    group_pk = Sequence(lambda n: n)


class UserFactory(alchemy.SQLAlchemyModelFactory):
    class Meta(object):
        model = User
        sqlalchemy_session = db.session  # the SQLAlchemy session object

    user_pk = Sequence(lambda n: n)
    name = Sequence(lambda n: u'User %d' % n)  # coding=utf-8
    group_fk = LazyAttribute(lambda a: GroupFactory().group_pk)


# ----------------------
#   Factory tests
# ----------------------
# Create a new Group from our factory
group_from_factory = GroupFactory(name='a new group name')
# Save it to our DB
db.session.add(group_from_factory)
db.session.commit()

# Verify that Group Saved correctly to DB
group_from_db = db.session.query(Group).filter(Group.group_pk == group_from_factory.group_pk).first()
assert group_from_db.group_pk is not None
assert group_from_db.name == 'a new group name'
assert group_from_db.group_pk == group_from_factory.group_pk

# Create a new User from our factory
user_from_factory = UserFactory()
db.session.add(user_from_factory)
# ----------------------------------------------
#   FAILS AT COMMIT() - NOT NULL constraint failed (group_fk is null)
# ----------------------------------------------
db.session.commit()

assert user_from_factory.user_pk is not None
assert user_from_factory.name is not None
assert user_from_factory.group_fk is not None

Candie