SQLAlchemy repository pattern example

crud.py

## Model classes


class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer,primary_key = True)
    book_id = Column(Integer,unique = True)
    title = Column(String, nullable = False)
    author = Column(String,index = True)
    created_date = Column(DateTime,server_default = func.now())
    updated_date = Column(DateTime,onupdate = func.now())
    last_update = Column(DateTime)
    introduction = Column(String)
    word_count = Column(Integer)
    publisher = Column(String,index = True)


class Volume(Base):
    __tablename__ = 'volume'
    id = Column(Integer,primary_key = True)
    book_id = Column(Integer,ForeignKey('books.id'),nullable = False,index = True)
    name = Column(String,nullable = False)

class VolumeDetails(Base):
    __tablename__ = 'volume_details'
    id = Column(Integer,primary_key = True)
    volume_id = Column(Integer,ForeignKey('volume.id'),nullable = False ,index = True)
    content = Column(String, nullable = False)

class SystemLog(Base):
    __tablename__ = 'sys_log'
    id = Column(Integer,primary_key = True)
    event_type = Column(String,nullable = False)
    log_level = Column(String,nullable = False)
    event_message = Column(String,nullable = False)
    event_time = Column(DateTime,server_default = func.now())
## Database utils for creating engine and session

class DatabaseUtil:
    DB_URL = 'sqlite:///data.db'    
    @classmethod
    @contextmanager
    def db_session(cls,db_url=None):
        """ Creates a context with an open SQLAlchemy session.
        """
        db_url = db_url if db_url else DatabaseUtil.DB_URL
        engine = create_engine(db_url, convert_unicode=True)
        db_session = scoped_session(sessionmaker(autocommit=False, autoflush=True, bind=engine))
        yield db_session
        db_session.close()


## Repository

class BookRepository:
    def persist(self,book):
        with open(DatabaseUtil.db_session()) as db:
            try:
                db.add(book)
                db.commit()
            except SQLAlchemyError as e:
                ## do logging
                db.rollback()

    def update(self,book_id,book):
         with open(DatabaseUtil.db_session()) as db:
            try:
                book = db.query(Book).filter(Book.book_id == book_id)\
                    .first()\
                book.title = 'modified title'
                db.commit()
            except SQLAlchemyError as e:
                ## do logging
                db.rollback()

    def delete(self,book):
        '''
        delete by object
        '''
        with open(DatabaseUtil.db_session()) as db:
            try:
                book = db.query(Book).filter(Book.book_id == book.book_id)\
                    .first()\
                book.delete()
                db.commit()
            except SQLAlchemyError as e:
                ## do logging
                db.rollback()