Setup The Newrelic Monitoring Agent On A Pylons App

Today I decide to signup for a free trial of the newrelic monitoring agent and I wanted to write on how to setup the agent on a Pylons app.
EDIT: Commenter Graham Dumpleton has advised against this setup in the comments. There’s the potential risk of the agent initializing twice and additional modules not working correctly. So please don’t use this setup on a production app.
Open the middleware.py file located on PRJNAME/config and add the following. Make sure to set the correct path where your newrelic.ini file is located

#import the agent
import newrelic.agent
#create a middleware class to initialize the agent
class NewRelicAgent(object):
    def __init__(self, app):
        self.app = app
        newrelic.agent.initialize('/path/to/the/file/newrelic.ini')
    @newrelic.agent.wsgi_application()
    def __call__(self, environ, start_response):
        return self.app(environ, start_response)

Now it’s time to add our custom middleware inside the make_app def

# CUSTOM MIDDLEWARE HERE (filtered by error handling middlewares)
app = NewRelicAgent(app)

And that’s it. Now it’s time to collect some stats :).

 

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()
psycopg2.extras.register_uuid()
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 helpers.py 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
      session.save()
      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."}
${h.text('captcha',maxlength=2)}

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':
        try:
          values = dict(request.params)
          schema = NewInquiry()
          results = schema.to_python(values)
        except Invalid, e:
          #raise error if something went wrong
        else:
          #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

https://code.google.com/p/python-twitter

  python setup.py build
  python setup.py install

Download and install dependencies

https://github.com/simplegeo/python-oauth2
http://code.google.com/p/httplib2/
http://cheeseshop.python.org/pypi/simplejson
Now you’re almost there. There’s a file called get_access_token.py 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 twittertoken.py.

import os
import sys
# parse_qsl moved to urlparse module in v2.6
try:
  from urlparse import parse_qsl
except:
  from cgi import parse_qsl
import oauth2 as oauth
REQUEST_TOKEN_URL = 'https://api.twitter.com/oauth/request_token'
ACCESS_TOKEN_URL  = 'https://api.twitter.com/oauth/access_token'
AUTHORIZATION_URL = 'https://api.twitter.com/oauth/authorize'
SIGNIN_URL        = 'https://api.twitter.com/oauth/authenticate'
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')
    else:
      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')
    else:
      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)
      token.set_verifier(oauth_verifier)
      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']) )
      else:
        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'''
  @property
  def message(self):
    '''Returns the first argument used to construct this error.'''
    return self.args[0]

Usage

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']
    session.save()
    # 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():
    log.debug(u.name)

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

 

Pylons and Facebook based Authorization with OAuth 2.0

Authorizing a user using Facebook OAuth 2.0 based login it’s pretty straighfoard. This is for the person who asked a question on stackoverflow.com I hope it helps.

Add the apikey, appid and secret of my Facebook app to the development.ini file

facebook.apikey = 42bc5a4051b274ede5cb73a6cd6fad56
facebook.secret = a355e66e8c0293390896a170f9d5r4c3
facebook.appid = 20894420580890

Configure my model/init.py file:

from PRJNAME.model.meta import Session, Base
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Unicode
from sqlalchemy import Table
from sqlalchemy import Text
from sqlalchemy import Unicode
from sqlalchemy import BigInteger
from sqlalchemy import TIMESTAMP
from sqlalchemy import ForeignKey,ForeignKeyConstraint
from sqlalchemy.orm import mapper
import datetime
def init_model(engine):
    """Call me before using any of the tables or classes in the model"""
    Session.configure(bind=engine,autoflush=True,autocommit=False)
def now():
    return datetime.datetime.now()
social_user = Table('social_user', Base.metadata,
    Column('id', Integer, primary_key=True),
    Column('oauth_provider', Text(), nullable=False),
    Column('oauth_id', Text() , nullable=False),
    Column('username', Unicode(50), nullable=False),
    Column('access_token', Unicode(200)),
    Column('created', TIMESTAMP(timezone=True), default=now()),
    )
  class Social(object):
    pass
  mapper(Social,social_user)

Download the Facebook Python-SDK

https://github.com/facebook/python-sdk

And save it on your project lib directory

Create a controller named account

~$ paster controller account

Add a view named login and render a login button

I’ll use https://graph.facebook.com/oauth/authorize as the url to start the authorization process

 import PRJNAME.lib.facebook as facebook
 import simplejson
 from pylons import config
 class AccountController(BaseController):
   def login(self):
     params = {
       'client_id':config['facebook.appid'],
       'scope' : 'offline_access,publish_stream,email', #permissions needed by the app
       'redirect_uri' : 'http://'+request.environ['HTTP_HOST']+'/account/auth'
     }
     keys = params.keys()[:]
     keys.sort()
     q_str = unicode()
     for name in keys:
       q_str+= ("&%s=%s" % (name,params[name]))
     c.login_url = 'https://graph.facebook.com/oauth/authorize?' + q_str
     return render('/website/login.html')

login.html

Login with Facebook

login with Facebook

Add a function on helpers.py

This function will create our access_token from our code request params

  import urllib
  import json
  def getAccessTokenFromCode(applicationId, applicationSecret, code,redirect_uri):
    q_str = unicode()
    params = {
      'client_id':applicationId,
      'client_secret':applicationSecret,
      'code':code,
      'redirect_uri':'http://'+request.environ['HTTP_HOST']+redirect_uri
    }
    keys = params.keys()[:]
    keys.sort()
    for name in keys:
      q_str+= ("&%s=%s" % (name, params[name]))
    url = 'https://graph.facebook.com/oauth/access_token?'+ q_str
    response = urllib.urlopen(url).read()
    result = response.split('=')[-1]
    if result:
      return result

Create an auth view on account controller

I do this because in the link button there is an argument called redirect_uri that points to http://ww.example.com/account/auth.

from PRJNAME.lib.helpers import getAccessTokenFromCode
from pylons import config
def auth(self):
if request.params.has_key('code'):
      access_token = getAccessTokenFromCode(
       config['facebook.appid'],
       config['facebook.secret'],
       request.params.get('code'),
       redirect_uri='/account/auth'
      )
      graph = facebook.GraphAPI(access_token)
      #grab the user data
      fbdata = graph.get_object("me")
      #query the db to check if user is already authorized
      q = Session.query(model.Social)
      user = q.filter_by(oauth_id=fbdata['id']).first()
      if user:
       # you can redirect to a dashboard
       return 'already in db and authorized'
      else:
      #insert user data to db
      fb = model.Social()
      fb.oauth_provider = 'Facebook'
      fb.oauth_id = fbdata['id']
      fb.username = fbdata['name']
      fb.access_token = access_token
      Session.add(fb)
      Session.commit()
      # redirect to dashboard page

And that’s pretty much it.

 

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
### BEGIN INIT INFO
# 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
### END INIT INFO
PROJECT=/var/www/ProjectName
PID_DIR=/opt/nginx/logs/ProjectName/
PID_FILE=/opt/nginx/logs/ProjectName/paster.pid
LOG_FILE=/opt/nginx/logs/ProjectName/paster.log
USER=root
GROUP=root
PROD_FILE=/var/www/ProjectName/production.ini
RET_VAL=0
cd $PROJECT
case "$1" in
start)
virtual-env/bin/paster  serve \
--daemon \
--pid-file=$PID_FILE \
--log-file=$LOG_FILE \
--user=$USER \
--group=$GROUP \
$PROD_FILE \
start
;;
stop)
virtual-env/bin/paster  serve \
--daemon \
--pid-file=$PID_FILE \
--log-file=$LOG_FILE \
--user=$USER \
--group=$GROUP \
$PROD_FILE \
stop
;;
restart)
virtual-env/bin/paster  serve \
--daemon \
--pid-file=$PID_FILE \
--log-file=$LOG_FILE \
--user=$USER \
--group=$GROUP \
$PROD_FILE \
restart
;;
*)
echo $"Usage: $0 {start|stop|restart}"
exit 1
esac

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
 

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),
    })
orm.mapper(Permission,permission_table,properties={
      'groups':orm.relation(Group, secondary=group_permission_table),
    })
orm.mapper(Group,group_table,properties={
      'permissions':orm.relation(Permission, secondary=group_permission_table),
      'users':orm.relation(User, secondary=user_group_table),
    })
orm.mapper(UGT,user_group_table)
orm.mapper(GPT,group_permission_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)
)
CREATE TABLE groups(
  id serial NOT NULL,
  gname character varying(50),
  CONSTRAINT groups_pkey PRIMARY KEY (id)
)
CREATE TABLE users(
id serial NOT NULL,
username character varying(50),
email character varying(50),
upassword character varying(150),
CONSTRAINT users_pkey PRIMARY KEY (id)
)

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

 

Nginx, uWSGI, Pylons and the double slash problem

So there I am, deploying my first Pylons based app and thinking on which WSGI interface to go with. At the end I decided to go with uWSGI it has been proven to be a beast at handling requests even on high loads.
I setup my Nginx conf file, fired up uWSGI and found out that all of my links had a double slash i.e //home/new. I couldn’t belive it all my hard work for nothing. After a couple of searches I found out a reply on the Nginx mailing list from Igor Sysoev saying the solution is to add the qualified param to all the url() calls.

url(controller='account', action='new', qualified=True)

Indeed that solved the problem, for a second I thought my chance for testing the performance of uWSGI was gone.
 

 

Posting to a page wall with the Facebook SDK in Python

I’ve been developing in Python for quite a few months now and the need to develop a Facebook app has arrived.
So in my app I had the need to post content to a Fan Page wall and after a couple of tests I noticed the content being posted on the Fan Page on behalf of the Page owner and not the page itself.
To fix that I had to make the following changes inside the if condition where it checks for the arguments being posted:

if post_args is not None:
  if post_args['page_token']:
    post_args["access_token"] = post_args['page_token']#self.access_token
  else:
    post_args["access_token"] = self.access_token
else:
  args["access_token"] = self.access_token

So now I’m checking for an extra argument in this case page_token, if it’s set we use that token instead of the token that was used to initialize the Facebook Graph object.
With this changes in place we add the extra parameter page_token to our attachment and it shoud work as expected, with the content being posted on behalf of the page.

graph = facebook.GraphAPI(str(access_token))
attach = {
  "name": 'Hello world',
  "link": 'http://www.example.com',
  "caption": 'test post',
  "description": 'some test',
  "picture" : 'http://www.example.com/picture.jpg'
  "page_token" : str(page_token)
}
msg = 'New content posted'
post = graph.put_wall_post(message=msg, attachment=attach,profile_id=str(page_id))
  if post:
    return 'posted'