Using SQLAlchemy and Django FormSet to work with an external database

Manipulating an external database with SQLAlchemy and Django FormSet is not as easy as Form (http://iambusychangingtheworld.blogspot.com/2013/07/django-make-use-of-django-forms.html). It's even more complicated when you want to do the CRUD. I had to do some hacks to achieve that.

For example, I have an external database table as following:

merge table:
=======================
|| id  || original     || translated     ||
=======================
|| 1   || course123 || courseABC ||
|| 2   || course456 || courseABC ||
|| 3   || course234 || courseABC ||
|| ...  || ...              || ...                 ||
=======================

1. First, define the merge table model with SQLAlchemy (MyProject.utils.merge_models):

class Merge(Base):
__tablename__ = 'merge'

id = Column(Integer, primary_key=True)
original = Column(String(100))
translated = Column(String(100))

def __init__(self, original, translated):
self.original = original
self.translated = translated

def __repr__(self):
return "<Merge(%s, %s)>" % (self.original, self.translated)


2. Next, write the database connector and utility methods to create, read, update, and delete (MyProject.utils.merge_utils):

from sqlalchemy import *
from sqlalchemy.orm import *
import os
from MyProject.utils.merge_models import Merge

DB_HOST = 'localhost'
DBMS = 'mysql'
DB_PORT = '3306'
DB_NAME = 'myexternaldb'
DB_USER = 'myuser'
DB_PWD = 'myuserpassword'

DB_ENGINE_URL = '%s://%s:%s@%s:%s/%s' % (DBMS, DB_USER, DB_PWD, DB_HOST,\
DB_PORT, DB_NAME)

class MyDBConnector(object):
engine = None
Session = None

def __init__(self):
self.connect()

def connect(self):
if self.engine is None:
self.engine = create_engine(DB_ENGINE_URL, echo=False, \
pool_size = 100, pool_recycle=3600)

if self.Session is None:
self.Session = scoped_session(sessionmaker(bind=self.engine,expire_on_commit=False))

def get_session(self):
self.connect()
return self.Session()

def close(self):
self.Session.remove()

def get_merges(db):
value = []
session = db.get_session()
try:
value = session.query(Merge).all()
except:
session.rollback()
raise
finally:
session.commit()
db.close()


return value

def update_or_create_merge(db, merge_id, org, trans):
session = db.get_session()
try: # get the merge
merge = session.query(Merge).get(merge_id)
except: # create a new merge
print "\nCannot get course merge %s\n" % (str(merge_id))
print "\nCreate new one\n"
if org and trans:
merge = Merge(original=org, translated=trans)
session.add(merge)
else:
print "Dont create null object"
else: # update the merge
merge.original = org
merge.translated = trans
finally:
session.commit()
db.close()

def delete_merge(db, merge_id):
session = db.get_session()
try:
merge = session.query(Merge).get(merge_id)
except:
print "\nMerge %s not found to delete\n" % str(merge_id)
else:
print "\nDeleting merge %s\n" % str(merge_id)
session.delete(merge)
finally:
session.commit()
db.close()

# get all merge data into a dictionary with specific keywords used for Django formset
def init_merge_data():
db = MyDBConnector()
merges = get_merges(db)
total = str(len(merges) + 1)

      # important!
data = {
'form-TOTAL_FORMS': total,
'form-INITIAL_FORMS': u'2',
'form-MAX_NUM_FORMS': u'',
}

for i in xrange(len(merges)):
data['form-' + str(i) + '-merge_id'] = merges[i].id
data['form-' + str(i) + '-c_org'] = merges[i].courseid_original
data['form-' + str(i) + '-c_trans'] = merges[i].courseid_translated


return data

Notes: the first three items of the data dictionary is important. They are used for the management form

3. Merge form and formset (MyProject.myapp.forms.py):

from django import forms
from django.forms.widgets import *
from django.forms.formsets import formset_factory
from MyProject.utils import merge_models, merge_utils

class MergeForm(forms.Form):
merge_id = forms.IntegerField(required=False)
org = forms.CharField(max_length=255, required=True, widget=forms.TextInput(attrs={'class' : 'mergefield'}))
trans = forms.CharField(max_length=255, required=True, widget=forms.TextInput(attrs={'class' : 'mergefield'}))

def save(self):
db = merge_utils.MyDBConnector()
if 'merge_id' in self.cleaned_data.keys() and self.cleaned_data['merge_id']:
merge_id = int(self.cleaned_data['merge_id'])
else:
merge_id = None
if 'org' in self.cleaned_data.keys():
org = self.cleaned_data['org']
else:
org = None
if 'trans' in self.cleaned_data.keys():
trans = self.cleaned_data['trans']
else:
trans = None

if 'DELETE' in self.cleaned_data.keys() and self.cleaned_data['DELETE']:
merge_utils.delete_merge(db, merge_id)
else:
merge_utils.update_or_create_merge(db, merge_id, org, trans)

MergeFormSet = formset_factory(MergeForm, can_delete=True)


Notes: It is important to set the can_delete=True if you want to delete a database row by the formset.


4. The views (MyProject.myapp.views):

from django.shortcuts import render_to_response, redirect
from django.template import RequestContext
from MyProject.utils.merge_utils import *
from MyProject.myapp.forms import *

@login_required
def merge(request):
if request.method == 'POST':
merges_formset = MergeFormSet(request.POST, request.FILES)
if merges_formset.is_valid():
for form in merges_formset:
if form.is_valid():
form.save()
return redirect('merge')
else:
data = init_merge_data()
merges_formset = MergeFormSet(data)

return render_to_response('merge.html', locals(), context_instance=RequestContext(request))


5. Finally, render the formset in the template (merge.html):

<form method="POST" action="">{% csrf_token %}
{{ merges_formset.management_form }}
<table>
<tr>
<th>Original</th>
<th>Translated</th>
<th>Delete?</th>
</tr>
{% for form in merges_formset %}
<tr>
{{ form.merge_id.as_hidden }}
<td>{{ form.org }}</td>
<td>{{ form.trans }}</td>
<td>
{% if merges_formset.can_delete %}
{% if not forloop.last %}
{{ form.DELETE }}
{% else %}
new merge
{% endif %}
{% endif %}
</td>
</tr>
{% endfor %}
</table>
<div>
<button type="submit">Submit</button>
</div>
</form>


This is genius!