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.


Running Standalone SQLAlchemy Scripts in Pyramid

From time to time there comes the need to run automated scripts managed by either Python or Cron. In a recent project I had to run a standalone script that checks records on a database at certain hours. So here’s what I came up with. Most of the script is based on this post by Daniel Mayer.

from paste.deploy import appconfig
from sqlalchemy import engine_from_config
from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound
#here is where sqlalchemy objects are imported
from PRJ.models import DBSession as db,DBRecords
#import the session manager. This way commits will be handled automatically by the Zope Transaction Manager
import transaction
# Load Application Configuration and Return as Dictionary
conf = appconfig('config:' + 'development.ini',
# Bind Engine Based on Config
engine = engine_from_config(conf, 'sqlalchemy.')
#Query the DB
data = db.query(DBRecords).one()
with transaction.manager:
  for record in data:
    #query or update DB

Facebook FQL Request Error

Today while attempting to query Events data through FQL, I received the following error:

Impersonated access tokens can only be used with the Graph API.

It was a strange error since I had already created my access token with the right scope. To query the data I’m using the pythonforfacebook SDK which uses as the url for FQL requests (REST API).
I then tried the graph method ( which worked perfectly, but the response didn’t include Event pictures and I really need pictures to be present on my events listings. So what worked for me was to make a GET request thought the Graph API instead of a POST.
I use Requests for this so the code is very simple.

data = db.query(Model) #retrieve my access_token from DB
payload = {'q' : query, 'access_token': data.access_token,'format' : 'json'}
res = requests.get('', params=payload)

With this change everything worked as expected.


Simpleform Localization in Pyramid

In a recent project I had to localize the errors thrown by the pyramid_simpleform package. Googling for information I couldn’t find how to do it, so here’s what worked for me at the end.

  from pyramid.i18n import get_locale_name
  from pyramid_simpleform import Form,State
  from formencode import api as formencode_api
  def includeme(config):
    config.add_route('login', '/login')
  def login(request):
     set the language in FormEncode according to the request url param _LOCALE_
     form = Form(request,
     set an empty gettext translation function,
     since FormEncode has one already
     configured in the set_stdtranslation function
     form.state._ = ''
     return dict(renderer=FormRenderer(form))

And that’s it, try it for example Make sure the action param in your form passes the _LOCALE_ value if the method is set to post.


Manage cron jobs with python-crontab

Cron is the main time based scheduler for any linux based system and is available in almost every distro. And in a recent project I had the task to manage jobs in cron from python. Searching for a good cron manager I came across python-crontab. It makes it really easy to manage jobs directly from cron, here are some examples:
NOTE: This examples used version 0.9.6, there’s a new version available 1.2 on pypi along with some examples, the main difference is that the API has been changed from slice calls to be properties instead of methods.

Installing python-crontab is easy as pie. First we install our virtual enviroment:

cd /var/www
python --no-site-packages prj-env
cd prj-env

Then we proceed to install python-crontab

pip install python-crontab
easy_install python-crontab

Let’s schedule a job to be executed everyday at 12pm

from crontab import CronTab
Here the object can take two parameters one for setting
the user cron jobs, it defaults to the current user
executing the script if ommited. The fake_tab parameter
sets a testing variable. So you can print what could be
written to the file onscreen instead or writting directly
into the crontab file.
tab = CronTab(user='www',fake_tab='True')
cmd = '/var/www/pjr-env/bin/python /var/www/PRJ/'
# You can even set a comment for this command
cron_job =, comment='This is the main command')
#writes content to crontab
print tab.render()

It will print out

0 12 * * * /var/www/pjr-env/bin/python /var/www/PRJ/

If we want to schedule a job to be executed every five minutes we could do something like this

from crontab import CronTab
tab = CronTab(user='www',fake_tab='True')
cmd = '/var/www/pjr-env/bin/python /var/www/PRJ/'
cron_job =
#writes content to crontab
print tab.render()

It will print out

*/5 * * * * /var/www/pjr-env/bin/python /var/www/PRJ/

If we want to schedule a job for a specific range of hours for example only working hours, we could do the following

from crontab import CronTab
tab = CronTab(user='www',fake_tab='True')
cmd = '/var/www/pjr-env/bin/python /var/www/PRJ/'
cron_job =
#writes content to crontab
print tab.render()

It will print out

0 09-18 * * * /var/www/pjr-env/bin/python /var/www/PRJ/

Now to schedule a job to run twice a day at 11 and 16 hrs, we could do the following

from crontab import CronTab
tab = CronTab(user='www',fake_tab='True')
cmd = '/var/www/pjr-env/bin/python /var/www/PRJ/'
cron_job =
#writes content to crontab
print tab.render()

it will print out

0 11,16 * * * /var/www/pjr-env/bin/python /var/www/PRJ/

Let’s delete the previous command

from crontab import CronTab
cmd = '/var/www/pjr-env/bin/python /var/www/PRJ/'
tab = CronTab(user='www',fake_tab='True')
cron_job = tab.find_command(cmd)
if len(cron_job) > 0:
#writes content to crontab

So there you have it, examples that make python-crontab a great python manager for cron jobs. Now I know there are pure Python implementations like this, an event scheduler named sched and libraries like Kronos. But I decided to keep things simple.


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.


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.