Random Musings of an Insane Mind

While working with Pyramid it was worth taking a good look at deform. During my brief look before, the one thing that appeared to be missing was a method to quickly integrate with SQLAlchemy backed data. For a handful of forms, manually creating the appstruct to be passed to the Form wouldn’t be difficult, but, for some more significant applications, it could be quite cumbersome. The following is a one page application that allows you to create a new user, or, edit an existing user by specifying the userid in the URL. The two pieces that do the real work are record_to_appstruct and merge_session_with_post. record_to_appstruct takes the class returned from SQLAlchemy and converts it to an appstruct that deform likes. Once we get the validated data, we merge the record with the post items using merge_session_with_post, merge the record and present a screen with the post data.

This is more a proof of concept, but, works well enough that we were able to convert quite a few forms and work more closely with Deform and Pyramid as we do more development.

__init.py__:

from pyramid.configuration import Configurator
from paste.deploy.converters import asbool

from defo.models import initialize_sql

def app(global_config, **settings):
    """ This function returns a WSGI application.
    
    It is usually called by the PasteDeploy framework during 
    ``paster serve``.
    """
    db_string = settings.get('db_string')
    if db_string is None:
        raise ValueError("No 'db_string' value in application configuration.")
    db_echo = settings.get('db_echo', 'false')
    initialize_sql(db_string, asbool(db_echo))
    config = Configurator(settings=settings)
    config.begin()
    config.add_static_view('static', 'defo:static')
    config.add_route('home', '/', view='defo.views.edit',
                     view_renderer='test.mako')
    config.add_route('homeid', '/:id', view='defo.views.edit',
                     view_renderer='test.mako')
    config.end()
    return config.make_wsgi_app()

views.py:

from defo.models import DBSession
from defo.models import MyModel

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import *
from sqlalchemy.databases import mysql
from sqlalchemy.orm import relation, backref, synonym
from sqlalchemy.orm.exc import NoResultFound

import deform
import colander
from webhelpers import constants

class UserSchema(colander.Schema):
    username = colander.SchemaNode(colander.String())
    contact = colander.SchemaNode(colander.String())
    email = colander.SchemaNode(colander.String())
    company = colander.SchemaNode(colander.String())
    addr1 = colander.SchemaNode(colander.String())
    addr2 = colander.SchemaNode(colander.String(), missing=u'',)
    city = colander.SchemaNode(colander.String())
    state = colander.SchemaNode(colander.String())
    zip = colander.SchemaNode(colander.String())
    country = colander.SchemaNode(
        colander.String(),
        widget = deform.widget.SelectWidget(values=constants.country_codes()),
    )
    phone = colander.SchemaNode(colander.String())

class AuthUser(Base):
    __tablename__ = 'auth_users'

    id = Column(mysql.BIGINT(20, unsigned=True), primary_key=True, autoincrement=True)
    username = Column(Unicode(80), nullable=False)
    email = Column(Unicode(80), nullable=False)
    contact = Column(Unicode(80), nullable=False)
    company = Column(Unicode(80), nullable=False)
    addr1 = Column(Unicode(80), nullable=False)
    addr2 = Column(Unicode(80))
    city = Column(Unicode(80), nullable=False)
    state = Column(Unicode(80), nullable=False)
    zip = Column(Unicode(80), nullable=False)
    country = Column(Unicode(80), nullable=False)
    phone = Column(Unicode(80), nullable=False)

def record_to_appstruct(self):
    return dict([(k, self.__dict__[k]) for k in sorted(self.__dict__) if '_sa_' != k[:4]])

def merge_session_with_post(session, post):
    for key,value in post:
        setattr(session, key, value)
    return session

def edit(request):
    dbsession = DBSession()
    if request.matchdict.has_key('id'):
        record = dbsession.query(AuthUser). \
                 filter_by(id=request.matchdict['id']).first()
    else:
        record = AuthUser()

    schema = UserSchema()
    form = deform.Form(schema, buttons=('submit',))
    if request.POST:
        try:
            appstruct = form.validate(request.POST.items())
        except deform.ValidationFailure, e:
            return {'form':e.render()}
        record = merge_session_with_post(record, request.POST.items())
        dbsession.merge(record)
        dbsession.flush()
        return {'formdata':appstruct}
    else:
        appstruct = record_to_appstruct(record)
    return {'form':form.render(appstruct=appstruct)}

test.mako:

<html>
<head>
  <title>
    Deform Demo Site
  </title>
  <!-- Meta Tags -->
  <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
  <!-- CSS -->
  <link rel="stylesheet" href="/static/css/form.css" type="text/css" />
  <link rel="stylesheet" href="/static/css/theme.css" type="text/css" />
  <!-- JavaScript -->
  <script type="text/javascript"
          src="/static/scripts/jquery-1.4.2.min.js"></script>
  <script type="text/javascript"
          src="/static/scripts/deform.js"></script>
</head>
<body>
this is a test template in mako
<p>
% if form:
  ${form|n}
% else:
We got our form data: ${formdata}
% endif
<script type="text/javascript">
   deform.load()
</script>
</body>
</html>

Tags: deform, pyramid, repoze, sqlalchemy

This entry was posted on Monday, November 8th, 2010 at 3:51 am and is filed under Framework. You can follow any responses to this entry through the RSS 2.0 feed. You can skip to the end and leave a response. Pinging is currently not allowed.