Real Time Notifications with Pyramid and ZeroMQ

Sooner or later I had to implement real time notifications on my latest project Floresta. The problem was that most of the code is synchronous. I think the ideal solution will be to run aiohttp on a separate instace or in a separate thread, but since I’m still on python 2.7 that was not possible. After researching on the subject I opted to enable gevent on my current uWSGI setup and add ZeroMQ to the equation.
So the first thing I did was install pyzmq, gevent and ZeroMQ. On mac is easy just use homebrew and pip. Make sure you install ZeroMQ version 4.1.5 or higher and pymzq 15.4.0 or higher.
On Debian I had to do the following:

Install libsodium first

git clone git://
cd libsodium
./configure && make check
sudo make install
sudo ldconfig

Then compile ZeroMQ

git clone
cd zeromq-4.1.X
./configure && make check
sudo make install
sudo ldconfig

On the uWSGI side I had to add these lines to my .ini file

gevent = 100
gevent-monkey-patch = true
enable-threads = true

Now the view

# coding=utf-8
from pyramid.response import Response
from pyramid.view import view_config, notfound_view_config,forbidden_view_config
from pyramid.httpexceptions import (
from sqlalchemy.orm.exc import NoResultFound,MultipleResultsFound
import logging
import os
import datetime
import time
import json
import threading
import  zmq
log = logging.getLogger(__name__)
sock = "ipc:///tmp/zmq.test"
context = zmq.Context()
pub_socket = context.socket(zmq.PUB)
pub_lock = threading.Lock()
def message_generator():
        socket2 = context.socket(zmq.SUB)
        #make sure to connect and not bind. You can only have one connect and multiple binds
        you can filter messages based on user id from a DB table as in
        socket2.setsockopt(zmq.SUBSCRIBE, userid)
        socket2.setsockopt(zmq.SUBSCRIBE, ')
    except zmq.error.ZMQError:
        console.log("socket already in use, try restarting it")
        # run forever
        while True:
                msg = socket2.recv(zmq.NOBLOCK)
                # break out of the loop
                if msg == "EXIT":
                    console.log("exiting process.")
                yield "data: %s\n\n" % json.dumps({"message": msg})
                console.log("sending message")
                console.log("nothing send..")
    except GeneratorExit:
#the url that streams the events to the browser
def _orders_events(request):
    headers = [("Content-Type", "text/event-stream"),
                       ("Cache-Control", "no-cache")]
    response = Response(headerlist=headers)
    response.app_iter = message_generator(userid)
    return response
#the url that publishes messages to the subscriber.
def push(request):
    msg = json.loads(request.body)["message"]
    with pub_lock:
    return Response()

The message_generator def will run forever inside your app. The way to stop it is to send a message with the string “Exit”.
Once you have this view in place, all you have to do is connect to the events url from javascript with the following code.

var source = null;
$(function() {
    source = new EventSource("/events");
    source.addEventListener("message", messageReceived, false);
    source.onerror = eventSourceErrorFunction;
    var eventSourceErrorFunction = function(event){
        if (event.eventPhase == EventSource.CLOSED) {
            console.log("Event Source Closed");
function messageReceived(event) {
console.log("message arrived: " + msg)

Send test messages from the Javascript console

        url: "/events/push",
        data: JSON.stringify({message: "Remote message"}),
        type: "post",
        success: function() {
            console.log("message sent!");

If you plan on placing Nginx in front of uWSGI don’t forget to turn off uwsgi_buffering and set the uwsgi_read_timeout to 300 otherwise the messages will get stuck.


Facebook SDK For Python Deprecated

Yesterday Facebook without a notice or warning decided it to delete their SDK github repo for Python. Today the repo is back online but with a notice saying that they have no plans to update it anymore. I’ve been seeing this coming since they stop adding new features awhile ago. So anyways there’s an alternative that will help us continue developing applications for Facebook even without their official support.
The move of deleting the repo really took me by surprise. All I have to say is that I really feel frustrated with this company, because they have abandon the Python comunity.


UUID Objects in Pylons with SQLAlchemy

Recently I had to generate and store uuid objects into a Postgres database, but to my surprise SQLAlchemy kept showing the following error:

sqlalchemy.exc.ProgrammingError: (ProgrammingError) can't adapt type 'UUID'

So the solution was that I had to change my field type from varchar to uuid on my Postgres database, import the psycopg2 extras functions and register the uuid type:

import psycopg2.extras
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy import MetaData,Column, Table, ForeignKey
engine = create_engine('postgresql://user:pass@localhost/db',echo=True)
metadata = MetaData(bind=engine)
hash_key = uuid.uuid4()
conn = engine.connect()
query = conn.execute("insert into tbl (uuid_col) values(%(uuid)s",uuid=hash_key)

And voila, values were inserted correctly on my database.


Implementing a text based captcha in Pylons

One of the must have elements on any html form as an anti-spam measure is a captcha, and the captcha technique that has worked pretty well for me is text based. So here’s how to implement one.

Create a controller and name it captcha

 ~$ paster controller captcha

Add this function to your file

def makeCaptcha(lang):
      label = {
          1 : 'one',
          2 : 'two',
          3 : 'three',
          4 : 'four',
          5 : 'five',
          6 : 'six',
          7 : 'seven',
          8 : 'eight',
          9 : 'nine',
          10 : 'ten'
      num1 = random.randint(1,10)
      num2 = random.randint(1,10)
      res = num1 + num2
      session['captcha'] = res
      return [label[num1],label[num2]]

On the templates folder create an html file with an example form

<% capt = h.makeCaptcha() c.captcha = u"What is the result of the sum %s and %s?." % (capt[0],capt[1]) %> ${u"As an anti spam measure, please answer the following math question."}

Save it and call it captcha.html

Add the following to captcha controller

I decided to create a validation schema using formencode:

import formencode
from formencode import variabledecode
from formencode import validators
from formencode.validators import Invalid, FancyValidator
#this class will validate the captcha value entered by the user
class CaptchaValidator(formencode.FancyValidator):
    def _to_python(self,values,state):
        if session.get('captcha') != int(values.get('captcha')):
          raise formencode.Invalid(u"The math answer is incorrect",values, state)
          return values
class NewInquiry(formencode.Schema):
  allow_extra_fields = True
  filter_extra_field = True
  captcha = formencode.validators.String(
    not_empty = True,
    messages = {
      'empty' : 'You need to answer the math question.'
  # chain the captcha validator
  chained_validators = [CaptchaValidator()]
# our contact view
def contact(self):
      if request.method == 'POST':
          values = dict(request.params)
          schema = NewInquiry()
          results = schema.to_python(values)
        except Invalid, e:
          #raise error if something went wrong
          #send to contacts
      return render('captcha.html')

And that’s pretty much it. The only drawback I can find is that isn’t very intuitive at times since users my try to answer with words instead of numeric values. That’s why I added a maxlength of 2 characters to the input field.
But you could easily implement a Javascript validation function to notify the user to type numeric values instead of characters before she submits the form.


HTML lists with jQuery for mobile devices

Recently I had to transform a group of li elements into an accordion for a mobile version of a client website. So this is what I came up with.
I had an html that looked like this

  • Category1
    • Subcategory1
    • Subcategory2
  • Category2
    • Subcategory
    • Subcategory1

And I wanted to collapse all the subcategories on load leaving only the categories visible and when they click/touched each category subcategories will expand. So I wrote this script based off from Sam Croft script.

  $.fn.accordion = function() {
    var el = this;
    var catHeight;
    catHeight = new Array();
    el.find('.category ul').each(function(i){
      var category = $(this);
      catHeight[i] = category.height();
      el.find('.category-header ').bind('touchstart', function(e) {
        var toExpand = $(this).next('ul');
        var i = toExpand.index('li ul');
        if (toExpand.attr('id') == 'active') {
        } else {
        var active = $('#active')
        if (active.length > 0) {
        .attr('id', 'active')
        .css('height', catHeight[i]+'px')

And with this script I had to modify my html lists to look like this

My css file

.closed {  height: 0; display:none;}


$(document).ready(function() {

You can see a working example here.


Pylons and Twitter Based Authorization

Implementing Twitter authorization in Pylons is easy as butter, I will show you how to do it in the next steps.

Download python-twitter and install it

  python build
  python install

Download and install dependencies
Now you’re almost there. There’s a file called that you need to call in order to obtain the Twitter session secret_token, but I decided to create a class based from that file instead. Save this class on your project lib folder as

import os
import sys
# parse_qsl moved to urlparse module in v2.6
  from urlparse import parse_qsl
  from cgi import parse_qsl
import oauth2 as oauth
SIGNIN_URL        = ''
class GenerateToken(object):
  def __init__(self,consumer_key=None,consumer_secret=None):
    if consumer_key is None or consumer_secret is None:
      raise TokenError('please add a consumer key and secret')
      signature_method_hmac_sha1 = oauth.SignatureMethod_HMAC_SHA1()
      self.oauth_consumer             = oauth.Consumer(key=consumer_key, secret=consumer_secret)
      self.oauth_client               = oauth.Client(self.oauth_consumer)
  def getrequestTokenURL(self):
    resp, content = self.oauth_client.request(REQUEST_TOKEN_URL, 'GET')
    if resp['status'] != '200':
      raise TokenError('Invalid response from Twitter')
      request_token = dict(parse_qsl(content))
      pieces = {
        'url' : "%s?oauth_token=%s" % (AUTHORIZATION_URL, request_token['oauth_token']),
        'token_secret': request_token['oauth_token_secret']
      return pieces
  def authRequest(self,oauth_token=None,oauth_token_secret=None,oauth_verifier=None):
      token = oauth.Token(oauth_token,oauth_token_secret)
      oauth_client  = oauth.Client(self.oauth_consumer, token)
      resp, content = oauth_client.request(ACCESS_TOKEN_URL, method='POST', body='oauth_verifier=%s' % oauth_verifier)
      access_token  = dict(parse_qsl(content))
      if resp['status'] != '200':
        raise TokenError('The request for a Token %s did not succeed: %s' % (access_token,resp['status']) )
        auth = {
          'access_token' : access_token['oauth_token'],
          'access_token_secret' : access_token['oauth_token_secret'],
        return auth
class TokenError(Exception):
  '''Base class for Token errors'''
  def message(self):
    '''Returns the first argument used to construct this error.'''
    return self.args[0]


create a controller in this case I’ll create an account controller

 paster controller account
import twitter
import PRJNAME.lib.twittertoken as twittertoken
def twitter_auth(self):
    token = twittertoken.GenerateToken(consumer_key=config['twitter.key'], consumer_secret=config['twitter.secret'])
    request_token = token.getrequestTokenURL()
    #save the token secret it will be used to generate the user's access_token and and access_secret
    session['token_secret'] = request_token['token_secret']
    # redirect to twitter screen
    return redirect(url(request_token['url']))
def twitter_preferences(self):
   params = request.params
   twittertoken.GenerateToken(consumer_key=config['twitter.key'], consumer_secret=config['twitter.secret'])
  auth = twittertoken.authRequest(oauth_token=params.get('oauth_token'),oauth_token_secret=session.get('token_secret'),oauth_verifier=params.get('oauth_verifier'))
  if auth['access_token'] and auth['access_token_secret']:
  #save to db or get user friend list
  for u in api.GetFriends():

I hope it helps to someone looking to implement Twitter on their Pylons projects.


Android SDK and Local Testing

It’s been awhile since I installed the Android SDK, today I’m starting a new android app so I updated to the latest version. To my surprise the tools have moved to different folders so just as I was ready to start testing locally I couldn’t find adb (adroid debugger bridge) in the tools folder, turns out it’s now located in the plataform-tools folder.
Here are the steps I’m using:
First you have to create an Android Virtual Device. Star the Android SDK Manager

android:~$ cd tools
tools:~$ ./android

Download the Android API you need or download them all
download android api
Create your Android Virtual Device
create android virtual device
Now you could start you virtual device from the SDK Manager but I prefer to do it from the shell since I’m able to add command options like

  • -wipe-data (To erase data stored from previous sessions)
  • -shell (So I can edit the /etc/hosts file later)
  • -no-boot-anim (It helps the emulator to boot up faster)
  • -noskin (It minimizes memory consumption)
  tools:~$ ./emulator -avd test -wipe-data -shell -no-boot-anim -noskin

Open a new shell tab and go to the platform-tools folder and remount the image with the Android Debugger Bridge. This is to get rid of the file write permissions issue.

$ cd platform-tools
platform-tools:~$./adb remount

Now go back to the tab where we started out emulator and edit the hosts file to point to a local domain. It’s important to mention that the /etc/hosts file in the emulator is different from then system /etc/hosts. Replace to your desired local domain

tools:~$ echo '' >> /etc/hosts

And that’s it. Now you can browse locally. Hope it helps someone.


Paste Server init script in Debian

This past few months I’ve set up a few beta projects on a single Debian server, and now it has become a tedious task typing the command to start|restart|stop paste for each project everytime I push new code. So I decide it to create a init script to simplify things.
For this I’ve my virtual environment installed inside the project folder. Make sure to replace virtual-env whit the name of your virtual environment folder and ProjectName with the name of your project folder. Here’s how it looks:

#! /bin/sh
# Provides:          paster
# Required-Start:    $all
# Required-Stop:     $all
# Default-Start:     2 3 4 5
# Default-Stop:      0 1 6
# Short-Description: starts the paster server
# Description:       starts paster
case "$1" in
virtual-env/bin/paster  serve \
--daemon \
--pid-file=$PID_FILE \
--log-file=$LOG_FILE \
--user=$USER \
--group=$GROUP \
virtual-env/bin/paster  serve \
--daemon \
--pid-file=$PID_FILE \
--log-file=$LOG_FILE \
--user=$USER \
--group=$GROUP \
virtual-env/bin/paster  serve \
--daemon \
--pid-file=$PID_FILE \
--log-file=$LOG_FILE \
--user=$USER \
--group=$GROUP \
echo $"Usage: $0 {start|stop|restart}"
exit 1

So now I can save this file as projectname-paster and move it to /etc/init.d/ and start it with:

sudo /etc/init.d/projectname-paster start | restart | stop

Host Email with Postfix, Dovecot and Postgres on Debian 6 (Squeeze)

Recently I had to configure a web server from scratch, which I haven’t done in a while. So I took a look at the guides from my current hosting provider Linode. They have a guide covering debian 5 (Lenny) which is basically the same for debian 6 (squeeze). To make this a bit more exciting I decided to replace MySQL for Postgres since I really didn’t need it. So I will only post about the changes you need to perform based from the Linode guide.

Install Required Packages

Replace the mysql related packages for postgres and replace mailx for bsd-mailx.

apt-get install postfix postfix-pgsql postfix-doc dovecot-common dovecot-imapd dovecot-pop3d postfix-tls libsasl2-2 libsasl2-modules libsasl2-modules-sql sasl2-bin libpam-pgsql openssl telnet bsd-mailx postgresql-contrib postgresql-8.4

Set up Postgres for Virtual Domains and Users

Open /etc/postgres/8.4/main/pg_hba.conf and make sure local connections require a password

host    all         all          md5

You could also make all local connections trusted and in this case no password will be needed, but of course it could represent a security risk

host    all         all          trust

This next step is optional but as a security measure I like to have Postgres listen on a diferent port other than the default.
Open /etc/postgres/8.4/main/postgres.conf and uncomment the following line and add 6789 as the port number

port = 6789

Create the user for the mail database.

sudo -u postgres createuser  -P -E -s -d mail_admin

Now create the database.

sudo -u postgres createdb -O mail_admin mail -E UTF-8

Grant all privileges.


Create domains, forwardings, users and transport tables.

CREATE TABLE domains (domain varchar(50) NOT NULL, PRIMARY KEY (domain) );
CREATE TABLE forwardings (source varchar(80) NOT NULL, destination TEXT NOT NULL, PRIMARY KEY (source) );
CREATE TABLE users (email varchar(80) NOT NULL, password varchar(20) NOT NULL, PRIMARY KEY (email) );
CREATE TABLE transport ( domain varchar(128) NOT NULL default '', transport varchar(128) NOT NULL default '', constraint fieldomainunique UNIQUE  (domain) );

Now I’ll create a domain and user in mail database. But first I will use Postgres pgcrypto functions since passwords need to be encrypted using the UNIX original encryption scheme. Replace your_password with your own password.

mail=#\i /usr/share/postgresql/8.4/contrib/pgcrypto.sql
INSERT INTO domains (domain) VALUES ('');
INSERT INTO users (email, password) VALUES ('', crypt('your_password', gen_salt('des') ));

Configure Postfix to work with Postgres

Create file /etc/postfix/

user = mail_admin
password = mail_admin_password
dbname = mail
query = SELECT domain AS virtual FROM domains WHERE domain='%s'
hosts =

Create file /etc/postfix/

user = mail_admin
password = mail_admin_password
dbname = mail
query = SELECT destination FROM forwardings WHERE source='%s'
hosts =

Create file /etc/postfix/

user = mail_admin
password = mail_admin_password
dbname = mail
query = SELECT split_part(email, '@', 2) || '/' || split_part(email,'@', 1) FROM users WHERE email='%s';
hosts =

Create file /etc/postfix/

user = mail_admin
password = mail_admin_password
dbname = mail
query = SELECT email FROM users WHERE email='%s'
hosts =

Set the proper permissions

chmod o= /etc/postfix/pgsql-virtual_*.cf
chgrp postfix /etc/postfix/pgsql-virtual_*.cf

Add the following one line at a time. Replace with your hostname

postconf -e 'myhostname ='
postconf -e 'mydestination =, localhost, localhost.localdomain'
postconf -e 'mynetworks ='
postconf -e 'message_size_limit = 30720000'
postconf -e 'virtual_alias_domains ='
postconf -e 'virtual_alias_maps = proxy:pgsql:/etc/postfix/, pgsql:/etc/postfix/'
postconf -e 'virtual_mailbox_domains = proxy:pgsql:/etc/postfix/'
postconf -e 'virtual_mailbox_maps = proxy:pgsql:/etc/postfix/'
postconf -e 'virtual_mailbox_base = /home/vmail'postconf -e 'virtual_uid_maps = static:5000'postconf -e 'virtual_gid_maps = static:5000'
postconf -e 'smtpd_sasl_auth_enable = yes'postconf -e 'broken_sasl_auth_clients = yes'
postconf -e 'smtpd_sasl_authenticated_header = yes'
postconf -e 'smtpd_recipient_restrictions = permit_mynetworks, permit_sasl_authenticated, reject_unauth_destination'
postconf -e 'smtpd_use_tls = yes'
postconf -e 'smtpd_tls_cert_file = /etc/postfix/smtpd.cert'
postconf -e 'smtpd_tls_key_file = /etc/postfix/smtpd.key'
postconf -e 'virtual_create_maildirsize = yes'
postconf -e 'virtual_maildir_extended = yes'
postconf -e 'proxy_read_maps = $local_recipient_maps $mydestination $virtual_alias_maps $virtual_alias_domains $virtual_mailbox_maps $virtual_mailbox_domains $relay_recipient_maps $relay_domains $canonical_maps $sender_canonical_maps $recipient_canonical_maps $relocated_maps $transport_maps $mynetworks $virtual_mailbox_limit_maps'
postconf -e virtual_transport=dovecotpostconf -e dovecot_destination_recipient_limit=1

Change postfix listening port to 2525 in /etc/postfix/

2525      inet  n       -       -       -       -       smtpd

Or you could just uncomment the submission line and default to port 587.

submission inet n       -       -       -       -       smtpd

Configure saslauthd to use Postgres

Create file /etc/pam.d/smtp. Replace your_password with your own password.

auth    required user=mail_admin passwd=your_password host= port=6789 db=mail table=users usercolumn=email passwdcolumn=password crypt=1
account sufficient user=mail_admin passwd=your_password host= port=6789 db=mail table=users usercolumn=email passwdcolumn=password crypt=1

Create file /etc/postfix/sasl/smtpd.conf. Replace your_password with your own password.

pwcheck_method: saslauthd
mech_list: plain login
allow_plaintext: true
auxprop_plugin: sql
sql_engine: pgsql
sql_user: mail_admin
sql_passwd: your_password
sql_database: mail
sql_select: select password from users where email = '%u'

Create file /etc/pam_pgsql.conf. Replace your_password with your own password.

database = mail
host = localhost
user = mail_admin
password = your_password
table = users
user_column = email
pwd_column = password
pw_type = crypt

Set proper permissions to files.

chmod o= /etc/pam_pgsql.conf
chmod o= /etc/postfix/sasl/smtpd.conf

Configure Dovecot

One thing that was pointed out at the end of the guide by user Nathan  is to change cmusieve for sieve on file /etc/dovecot/dovecot.conf on this following part

protocol lda {
    log_path = /home/vmail/dovecot-deliver.log
    auth_socket_path = /var/run/dovecot/auth-master
    postmaster_address =
    mail_plugins = cmusieve #change to sieve
    global_script_path = /home/vmail/globalsieverc

Create file /etc/dovecot/dovecot-sql.conf. Replace your_password with your own password.

driver = pgsql
connect = host= port=6789 dbname=mail user=mail_admin password=your_password
default_pass_scheme = CRYPT
password_query = SELECT email as user, password FROM users WHERE email='%u';

And thats about it. If you managed to configure everything correctly congratulations!, if not post a comment and we’ll try to find the solution.


Remember that /var/log/syslog and /var/log/mail.log are your friends. You can also add the word debug at the end of each line on /etc/pam.d/smtp to help you troubleshoot any problems you may encounter.


repoze.what with reflected tables using SQLAlchemy in Pylons

One of the many things I like about SQLAlchemy is the feature of reflected tables, this means I can develop dashboards for existing applications with my current favorite framework(Pylons). So I had to create an authorization mechanism for my dashboard and this recipe from the Pylons cookbook was just what I needed.
The only thing I had to change was the way my tables were being mapped since the users table schema was already defined. This is what I did:

user_table = schema.Table('users', Base.metadata, autoload=True, autoload_with=engine)
user_group_table = schema.Table('user_groups',Base.metadata,autoload=True,autoload_with=engine)
group_permission_table = schema.Table('group_permission',Base.metadata,autoload=True,autoload_with=engine)
permission_table = schema.Table('permissions', Base.metadata,autoload=True,autoload_with=engine)
group_table = schema.Table('groups', Base.metadata,autoload=True,autoload_with=engine)
orm.mapper(User,user_table, properties={
      'groups':orm.relation(Group, secondary=user_group_table),
      'groups':orm.relation(Group, secondary=group_permission_table),
      'permissions':orm.relation(Permission, secondary=group_permission_table),
      'users':orm.relation(User, secondary=user_group_table),

And here’s the schema I had to create, except for the users table:

CREATE TABLE user_groups( 
id serial NOT NULL,  
user_id integer,  
group_id integer,  
CONSTRAINT user_groups_pkey PRIMARY KEY (id), 
CONSTRAINT gid_fkey FOREIGN KEY (group_id) REFERENCES groups (id)
CONSTRAINT uid_fkey FOREIGN KEY (user_id)   REFERENCES users (id)
CREATE TABLE group_permission(
id serial NOT NULL,
gid integer,
permid integer,
CONSTRAINT group_permission_pkey PRIMARY KEY (id),
CONSTRAINT groupid_fkey FOREIGN KEY (gid) REFERENCES groups (id),
CONSTRAINT permid_fkey FOREIGN KEY (permid) REFERENCES permissions (id)
CREATE TABLE permissions(
  id serial NOT NULL,
  pname character varying(50),
  CONSTRAINT permissions_pkey PRIMARY KEY (id)
  id serial NOT NULL,
  gname character varying(50),
  CONSTRAINT groups_pkey PRIMARY KEY (id)
id serial NOT NULL,
username character varying(50),
email character varying(50),
upassword character varying(150),

And that’s all there’s to it. I was very surprise on how flexible SQLAlchemy is.