Correct pattern for inserting row in SQLAlchemy

Question

I have an insert action that is kinda complex, it boils to:

• input: list of tags, book name
• INSERT INTO books (book_name) VALUES (book name)
• for each tag:
  • if tag does not exist:
    • INSERT INTO tags (tag_name) VALUES (tag name)
  • INSERT INTO books_tags (book_id, tag_id) VALUES (book id, tag id)

Can I do it somehow easier than just writing those in SQLALchemy syntax?

Answer 1

Short Answer: Use custom creator in the definition of association_proxy:

def _tag_find_or_create(tag_name):
    tag = Tag.query.filter_by(tag_name=tag_name).first()
    return tag or Tag(tag_name=tag_name)


class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True)
    book_name = Column(String)

    # relationship
    _tags = relationship('Tag', secondary='book_tag')
    tags = association_proxy('_tags', 'tag_name', creator=_tag_find_or_create)

---

Long Answer: The code below is a self-contained working example. Few words on the implementation:

• We use Association Proxy in order to simplify many-to-many relationship. Give this whole page a good read.
• In particular, we redefine the default creator function, which will first query the database/session to check for the Tag, and create one if not found.

Code:

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy import UniqueConstraint, ForeignKey
from sqlalchemy.orm import relationship, scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.associationproxy import association_proxy

# Configure test data SA
engine = create_engine('sqlite:///:memory:', echo=True)
session = scoped_session(sessionmaker(bind=engine))
Base = declarative_base(engine)
Base.query = session.query_property()


def _tag_find_or_create(tag_name):
    tag = Tag.query.filter_by(tag_name=tag_name).first()
    return tag or Tag(tag_name=tag_name)


class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True)
    book_name = Column(String)

    # relationship
    _tags = relationship('Tag', secondary='book_tag')
    tags = association_proxy('_tags', 'tag_name', creator=_tag_find_or_create)


class BookTag(Base):
    __tablename__ = 'book_tag'
    __tableargs__ = (UniqueConstraint('book_id', 'tag_id', name='book_tag_uc'),)

    id = Column(Integer, primary_key=True)
    book_id = Column(Integer, ForeignKey('book.id'))
    tag_id = Column(Integer, ForeignKey('tag.id'))


class Tag(Base):
    __tablename__ = 'tag'
    id = Column(Integer, primary_key=True)
    tag_name = Column(String, unique=True)


# CREATE SCHEMA
Base.metadata.create_all()


def _insert_test_data():
    book = Book(book_name="book-1")
    book.tags.append("fiction")
    book.tags.append("history")
    session.add(book)
    session.commit()

    assert 1 == len(Book.query.all())
    assert 2 == len(Tag.query.all())

    book2 = Book(book_name="book-2")
    book2.tags.append("history")
    book2.tags.append("crime")
    session.add(book2)
    session.commit()

    assert 2 == len(Book.query.all())
    assert 3 == len(Tag.query.all())


def _add_new_book(book_name, tags):
    book = Book.query.filter(Book.book_name == book_name).first()
    assert not(book), "Book with name '{}' already found [{}]".format(
        book_name, book)

    book = Book(book_name=book_name)
    for tag in tags:
        book.tags.append(tag)
    session.add(book)
    session.commit()


if __name__ == '__main__':
    _insert_test_data()
    _add_new_book('SuperMan', ['fiction', 'romance'])