SQLAlchemy to/from PyQt Adapters

alchemical_model.py

#!/usr/bin/env python3
# © 2013 Mark Harviston, BSD License
# 2016: update to use PyQt5/Python3, some enhancements by Christian González
"""
Qt data models that bind to SQLAlchemy queries
"""
from PyQt5.QtWidgets import QMessageBox 
from PyQt5.QtSql import QSqlTableModel
from PyQt5.Qt import QVariant, Qt


class SqlAlchemyTableModel(QSqlTableModel):
    """A Qt Table Model that binds to a SQLAlchemy query

    Example:
    >>> model = AlchemicalTableModel(Session,
            Entity, 
            [('Name', Entity.name, "name", {"editable": True} )])
    >>> table = QTableView()
    >>> table.setModel(model)
    """

    def __init__(self, session, entity, columns):
        """Constructor for the model.
        
        @param session: The SQLAlchemy session object.
        @param entity: The entity class that represents the SQLAlchemy data object 
        @param columns: A list of column 4-tuples
          (header, sqlalchemy column, column name, extra parameters as dict)
          if the sqlalchemy column object is 'Entity.name', then column name
          should probably be 'name'.
          'Entity.name' is what will be used when setting data and sorting,
          'name' will be used to retrieve the data.
        """
        
        super().__init__()
        #TODO self.sort_data = None
        self.session = session
        self.fields = columns
        self.query = session.query
        self.entity = entity

        self.results = None
        self.count = None
        self.sort = None
        self.filter = None

        self.refresh()

    def headerData(self, col, orientation, role):
        if orientation == Qt.Horizontal and role == Qt.DisplayRole:
            return QVariant(self.fields[col][0])
        return QVariant()

    def setFilter(self, newFilter):
        """Sets or clears the newFilter.
        
        Clear the filter by setting newFilter to None
        """
        self.filter = newFilter
        self.refresh()

    def refresh(self):
        """Recalculates self.results and self.count"""

        self.layoutAboutToBeChanged.emit()

        q = self.session.query
        
        if self.sort is not None:
            order, col = self.sort
            col = self.fields[col][1]
            if order == Qt.DescendingOrder:
                col = col.desc()
        else:
            col = None
            
        if self.filter is not None:
            q = q.filter(self.filter)

        self.results = q(self.entity).all()
        self.count = len(self.results)
        self.layoutChanged.emit()

    def flags(self, index):
        _flags = Qt.ItemIsEnabled | Qt.ItemIsSelectable

        if self.sort is not None:
            order, col = self.sort

            if self.fields[col][3].get('dnd', False) and index.column() == col:

                _flags |= Qt.ItemIsDragEnabled | Qt.ItemIsDropEnabled

        if self.fields[index.column()][3].get('editable', False):
            _flags |= Qt.ItemIsEditable

        return _flags

    def supportedDropActions(self):
        return Qt.MoveAction

    def dropMimeData(self, data, action, row, col, parent):
        if action != Qt.MoveAction:
            return

        return False

    def rowCount(self, parent):
        return self.count or 0

    def columnCount(self, parent):
        return len(self.fields)

    def data(self, index, role):
        if not index.isValid():
            return QVariant()

        elif role not in (Qt.DisplayRole, Qt.EditRole):
            return QVariant()

        row = self.results[index.row()]
        name = self.fields[index.column()][2]

        return getattr(row, name)

    def setData(self, index, value, role=None):
        row = self.results[index.row()]
        name = self.fields[index.column()][2]

        try:
            setattr(row, name, value.toString())
            self.session.commit()
        except Exception as ex:
            # FIXME: data layer should not open GUI Messagebox!
            QMessageBox.critical(None, 'SQL Error', ex)
            return False
        else:
            self.dataChanged.emit(index, index)
            return True

    def sort(self, col, order):
        """Sort table by given column number."""
        if self.sort != (order, col):
            self.sort = order, col
            self.refresh()

example.py

#(leaves out sqlalchemy & PyQt boilerplate, will not run)
#Define SQL Alchemy model
from qvariantalchemy import String, Integer, Boolean
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

class Entity(Base):
	__tablename__ = 'entities'

	ent_id = Column(Integer, primary_key=True)
	name = Column(String)
	enabled = Column(Boolean)
    
#create QTable Model/View
from alchemical_model import AlchemicalTableModel
model = AlchemicalTableModel(
    Session, #FIXME pass in sqlalchemy session object
    Entity, #sql alchemy mapped object
    [ # list of column 4-tuples(header, sqlalchemy column, column name, extra parameters as dict
      # if the sqlalchemy column object is Entity.name, then column name should probably be name,
      # Entity.name is what will be used when setting data, and sorting, 'name' will be used to retrieve the data.
        ('Entity Name', Entity.name, 'name', {'editable': True}),
        ('Enabled', Entity.enabled, 'enabled', {}),
    ])

table = QTableView(parent)
table.setModel(model)

qvariantalchemy.py

#!/usr/bin/env python2
#-*- coding=utf-8 -*-
# © 2013 Mark Harviston, BSD License
from __future__ import absolute_import, unicode_literals, print_function
"""
SQLAlchemy types for dealing with QVariants & various QTypes (like QString)
"""

import datetime

from PyQt4.QtCore import QVariant
from sqlalchemy import types


def gen_process_bind_param(pytype, toqtype, self, value, dialect):
  if value is None:
		return None
	elif isinstance(value, QVariant):
		return pytype(toqtype(value))
	elif not isinstance(value, pytype):
		return pytype(value)
	else:
		return value


class Integer(types.TypeDecorator):
	impl = types.Integer

	def process_bind_param(self, value, dialect):
		return gen_process_bind_param(
			long, lambda value: value.toLongLong(),
			self, value, dialect)


class Boolean(types.TypeDecorator):
	impl = types.Boolean

	def process_bind_param(self, value, dialect):
		return gen_process_bind_param(
			bool, lambda value: value.toBool(),
			self, value, dialect)


class String(types.TypeDecorator):
	impl = types.Unicode

	def process_bind_param(self, value, dialect):
		return gen_process_bind_param(
			unicode, lambda value: value.toString(),
			self, value, dialect)


class Enum(types.TypeDecorator):
	impl = types.Enum

	def process_bind_param(self, value, dialect):
		return gen_process_bind_param(
			unicode, lambda value: value.toString(),
			self, value, dialect)


class DateTime(types.DateTime):
	impl = types.DateTime

	def process_bind_param(self, value, dialect):
		return gen_process_bind_param(
			datetime.datetime, lambda value: value.toDateTime(),
			self, value, dialect)