Model

There is an important thing I need to introduce first, SQLAlchemy. Not only recording into the database, transacion is also involved. If we take a look at a code template which is generated from a tool of Pyramid, there is the transaction that is different in my source code.

I start at declaring some fields.

Base = declarative_base()
class Todo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(512), nullable=False)
    created_at = Column(DateTime, server_default=text('NOW()'), nullable=False)
    done_at = Column(DateTime)
    priority = Column(Integer, default=5) # 1 => the most priority, 10 => not important now 
 
    def __init__(self, task, done_at=None, priority=1):
        super(Todo, self).__init__()
        self.task = task
        self.done_at = done_at
        self.priority = priority

There are the fields to identify some properties like id, what to do, when to be finished and priority.
Another confusing is transaction

DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))

This code creates a database connection that passes ZopeTransactionExtension to scoped_session. ZopeTransactionExtension commits the transaction after finishing the processing the request. If there is any exception is raised then it will rollback.

The problem is if we follow an example on the site, it creates a transaction with transacton.manager. When it finishes work in transaction manager, it will commit so there is nothing to be rolled back. To add DBSession.remove() is for removing the data after testing finishes. The example in this doc doesn’t need to remove because it stores the data in sqlite which is a memory database. It removes automatically after testing.

Also, there is anothing to be added to make testing goes well. Let’s start at SQLAlchemy’s ORM method. Whatever it does through a SQLAlchemy’s session, it is like storing executed command to a stack and then execute all of them after returning in the function in the controller but what I expect is after saving, it should be retrived the result at that moment. Doesn’t need to wait until function returns.
To do that, just flush

DBSession.flush()

After flushing then todo’s instance is set some values such as there is a new todo that there is no id, it is a creating in the database. After flushing, id will be set as the id in row in the database. So if no flushing then no id to be set, we don’t know if it save passed. If fail, it will raise an exception that we don’t what happen exactly. It is just an error message from the database that is not friendly to the user. To catch the exception is out of the code’s scope also means we can’t catch the exception if no flushing.

Let’s back to the code. It’s not funny to flush every time after updating or deleting so add some helper methods and add some validation methods that SQLAlchemy doesn’t have.

class AppBase(object):
    _errors = None
 
    def save(self):
        if self.is_valid:
            try:
                DBSession.add(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def update(self):
        if self.is_valid:
            try:
                DBSession.merge(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def delete(self):
        try:
            DBSession.delete(self)
            DBSession.flush()
        except IntegrityError: # still dont have an idea to test this failure
            return False
        return True

Below is the validation added by myself.

    @property
    def errors(self):
        return self._errors
 
    @property
    def is_valid(self):
        return not bool(self._errors)
 
    def validate(self, validator, key, value):
        if not self._errors:
            self._errors = {}
        try:
            validator.to_python(value)
        except Invalid as e:
            if not self._errors.get(key):
                self._errors[key] = []
            self._errors[key].append(str(e))

There is a hooked method named ‘validates’ which is a decorator. It is invoked on value assinging.

    @validates('task')
    def validate_task(self, key, value):
        # validate
        return value

Task is a declared feild in Todo class and this method is in Todo class too.
The code below show how to extend this class.

Base = declarative_base(cls=AppBase)

And here is the whole code in models.py

from sqlalchemy import (
    Column,
    Integer,
    Text,
    String,
    DateTime,
    Integer
    )
 
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.expression import text
from sqlalchemy.exc import IntegrityError
 
from sqlalchemy.orm import (
    scoped_session,
    sessionmaker,
    validates,
    )
 
from zope.sqlalchemy import ZopeTransactionExtension
 
from formencode import validators, Invalid
 
DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
#DBSession = scoped_session(sessionmaker())
 
class AppBase(object):
    _errors = None
 
    def save(self):
        if self.is_valid:
            try:
                DBSession.add(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def update(self):
        if self.is_valid:
            try:
                DBSession.merge(self)
                DBSession.flush()
                return True
            except IntegrityError:
                return False
        return False
 
    def delete(self):
        try:
            DBSession.delete(self)
            DBSession.flush()
        except IntegrityError: # still dont have an idea to test this failure
            return False
        return True
 
    @property
    def errors(self):
        return self._errors
 
    @property
    def is_valid(self):
        return not bool(self._errors)
 
    def validate(self, validator, key, value):
        if not self._errors:
            self._errors = {}
        try:
            validator.to_python(value)
        except Invalid as e:
            if not self._errors.get(key):
                self._errors[key] = []
            self._errors[key].append(str(e))
 
Base = declarative_base(cls=AppBase)
 
class Todo(Base):
    __tablename__ = 'todos'
    id = Column(Integer, primary_key=True)
    task = Column(String(512), nullable=False)
    created_at = Column(DateTime, server_default=text('NOW()'), nullable=False)
    done_at = Column(DateTime)
    priority = Column(Integer, default=5) # 1 => the most priority, 10 => not important now 
 
    def __init__(self, task, done_at=None, priority=1):
        super(Todo, self).__init__()
        self.task = task
        self.done_at = done_at
        self.priority = priority
 
    @validates('task')
    def validate_task(self, key, value):
        self.validate(validators.String(not_empty=True), key, value)
        return value
 
    @validates('priority')
    def validate_priority(self, key, value):
        self.validate(validators.Int(), key, value)
        return value

The validation helper I chosed is a href=”http://www.formencode.org/en/latest/” target=”_blank”>formencode. As in the example, just needs validators.String(not_empty=True) means not allowed empty value and validators.Int() which is needed only Int.

Controller

Python frameworks often use word ‘views’ instead controllers. I’m not used to it so I will call it ‘controller(s)’ then. Let’s create the controller pacakge and put todos.py follow the test in the previous post.

from todolist.controllers.todos import create

There is a tricky part with the template engine. Because a bug on the builtin template engine, ZPT(Chameleon). The issue is using macro with doctype cause an error so I changed to mako

Read

Here is for listing.

def get_todo_set():
    todos = DBSession.query(Todo).filter(Todo.done_at==None).order_by(Todo.priority.desc()).all()
    done_todos = DBSession.query(Todo).filter(Todo.done_at!=None).order_by(Todo.priority.desc()).all()
    return todos, done_todos
 
@view_config(route_name='todo_index', renderer='todos/index.mako')
def index(request):
    todos, done_todos = get_todo_set()
    return dict(todos=todos, done_todos=done_todos)

I add get_todo_set separately. It can be used anywhere. Then see the code in the testing. You will see

        self.assertEqual(len(response['todos']), 3)
        self.assertEqual(len(response['done_todos']), 1)
        # test ordering
        self.assertEqual(response['todos'][0].task, "First task")
        self.assertEqual(response['todos'][2].task, "Second task")

There is a checking to a response that is retrived back from the controller.
You can try seeing and compare the rest of the controlling mehtod with the testing.

@view_config, as a decorator set what to do. Paramerters in the code:

• route_name for linking URI to a function in the controller
• renderer for assign a template file. It can be either xml or json and it will render the content-type as it is assigned.
• request_method to assign the http request method to be received.
• xhr to allow only XmlHttpRequest

To test @view_config needs to be tested with functional test such as checking whether a status code is 200 or content-type is json.

Create

@view_config(route_name='todo_create', renderer='json', request_method='POST', xhr=True)
def create(request):
    todo = Todo(
        task=request.POST.get('task', None),
        priority=request.POST.get('priority', None)
    )
    if not todo.save():
        return {'errors': todo.errors}
    return {'id': todo.id, 'task': todo.task, 'priority':todo.priority, 'messages': '%s has been created' % todo.task }

In the code, it needs only ajax request and needs to ensure that definitely create a todo. As explained above, id will be set after creating. I like cheking like this instead of leave the exception raised because I can check every case can happen.

Update

@view_config(route_name='todo_update', renderer='json', request_method='POST', xhr=True)
def update(request):
    todo_id = request.matchdict['id']
    try:
        todo = DBSession.query(Todo).filter_by(id=todo_id).one()
    except NoResultFound:
        return {'errors': "No todo id: %s" % todo_id}
 
    todo.task=request.POST.get('task', None)
    todo.priority=request.POST.get('priority', None)
 
    if not todo.update():
        return {'errors': todo.errors}
 
    return {'task': todo.task, 'priority':todo.priority, 'messages': '%s has been updated' % todo.task}

The method to update is merge() by put an instance into DBSession, if id is already set then it will update. If not, it will create.

Delete

@view_config(route_name='todo_delete', renderer='json', request_method='POST', xhr=True)
def delete(request):
    todo_id = request.matchdict['id']
    try:
        todo = DBSession.query(Todo).filter_by(id=todo_id).one()
    except NoResultFound:
        return {'errors': "No todo id: %s" % todo_id}
    if not todo.delete():
        return {'errors': "%s can't be deleted" % todo.task}
 
    return {'id': todo.id, 'messages': '%s has been deleted' % todo.task}

And the last one is deleting with a couple of error messages.

Actually, I wanted to work the UI out based on ajax but it took too long and I wanted to show the testing so I left it.

Summary

I’m happy with integration test. The tool like nosetest for doing the code coverage includes unittest is more convenient or unittest can test cover all of I want. It might be I just started so some code might not look ok especially SQLAlchemy part.

You can Download source code or

hg clone https://bitbucket.org/punneng/pyramid-testfirst

I will find free time to try functional test.

After spending 2-3 weeks with Pyramid, I realized that it’s not too complicated and there is good documents. But there is something a bit hard if you’ve worked on Rails is choosing an ORM. I spent a bit long to search for the orm and I chosed SQLAlchemy. Actually, I don’t know any thing about SQLAlchemy. Just heard that it’s quite light, not too heavy like ActiveRecord. After an experiment, it’s not too bad. Just need to used to how to read the document. Need to know where what you’re looking for. As far as I’m a developer, I’ve never found any document starts with writing test. I’s quite impressive to me. I actually want BDD like RSpect but I’m a newbie. Still finding now.

I’ll create a simple todolist and write the test covers CRUD. My tests should be failed totally. Once I implemented, they would be passed all. In order to to start, let’s follow the installation then you’ll see 2 testing tools which are Unittest and Test coverage. Unittest is a Python builtin library(PyUnit) and test coverage is a combination of Coverage and nose to customize the testing files in the projects.

Before writing test cases, let’s see how the testing in Pyramid are… there are 3 kinds of tests.

• Unit test is to test the smallest unit like functions or variables.
• Integration test is a group of tests that tests the units as a component that the units work together.
• Functional test is like the integration test which runs the application literally.

My test cases

What I want to test..

• C – Create should post a request to a function and return back a dict that contains todo data.
• R – Read(also Edit and list) should get the data that can be a dict of todo or a list of the dict of todos.
• U – Update supposed to be put. I wanted to make it restfully but I’m a newbie. Let’s set it as the post request and return back the dict of updated todo.
• D – Delete should post(delete in restful) the id and return back the deleted id

What I concern about these test cases is what the test retrieves in the provided functions. It actually is the real http response to be retrieved like http status is 200 or content type is json but this kind of concern is the functional test. There is another server library to mock the http request and posts to the full stack of the application if you want to test.

Set up and Tear down

A little more introduction about unittest in Pyramid. There are set up and tear down method for putting the code to be run when the test starts and run after the test ends. For example, testing.setUp() is the method to prepare the application environment for testing or the group of code to initial the data for testing in the set up method. Also after finish testing, remove that data in the tear down method.

class TestIntegrationTodolist(unittest.TestCase):
    def setUp(self):
        self.config = testing.setUp()
        from sqlalchemy import create_engine
        #engine = create_engine('sqlite:///todolist.sqlite')
        engine = create_engine('postgresql://dev:dev@localhost:5432/todolist_test')
 
        DBSession.configure(bind=engine)
        Base.metadata.create_all(engine)
        # the transaction commited after added so rolling back doesnt work on Postgres
        #with transaction.manager:
        instances = (
            Todo(task='Second task', priority=0),
            Todo(task='Thrid task', priority=5),
            Todo(task='First task', priority=10),
            Todo(task="Done task", priority=5, done_at=datetime.now())
        )
        DBSession.add_all(instances)
 
    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

it starts with conneting to PosgreSQL then create the connection between Pyramid and SQLAlchemy. After that is the group of code to initial the data for test. In the end, just remove the data.

Read

To Read in my todolist, there is 1 task which is listing all of the todo. What I excpect from the test is after request, it should return a list of dict of todos.

    def test_index_pass(self):
        """ test_index_pass """
        from todolist.controllers.todos import index
        request = testing.DummyRequest()
        response = index(request)
        self.assertEqual(len(response['todos']), 3)
        self.assertEqual(len(response['done_todos']), 1)
        # test ordering
        self.assertEqual(response['todos'][0].task, "First task")
        self.assertEqual(response['todos'][2].task, "Second task")

There is an addiional done_todos which is a list of dict to contains the done todos. I intended to show them in another list in the same page. done_at is the key to specific which todos are done.

One facility method is testing.DummyRequest() to create a request as a parameter to send to the get function.

Create

    def test_create_pass(self):
        """ test_create_pass """
        from todolist.controllers.todos import create
        params = {'task':'New task', 'priority':1}
        request = testing.DummyRequest(params=params, post=params)
        response = create(request)
        self.assertTrue(response['id'])
        self.assertEqual(response['task'], params['task'])
        self.assertEqual(response['priority'], params['priority'])
 
    def test_create_fail(self):
        """ test_create_fail """
        from todolist.controllers.todos import create
        params = {'task':"", 'priority': "low"}
        request = testing.DummyRequest(params=params, post=params)
        response = create(request)
        self.assertTrue(response['errors'])
        self.assertEqual(len(response['errors']['priority']), 1)
        self.assertEqual(len(response['errors']['task']), 1)
        self.assertEqual(response['errors']['task'], ['Please enter a value'])
        self.assertEqual(response['errors']['priority'], ['Please enter an integer value'])

incase of create, I provided 2 cases, pass and fail. If fail then show the notified message.

To create the post request with DummyRequest is just put the post value as a parameter. If want to know what are the parameters, here is the doc. You’ll see how to create a session too.

Update

    def test_update_pass(self):
        """ test_update_pass """
        from todolist.controllers.todos import update
        params = {'task':'Updated task', 'priority':1}
        # provide the todo with id
        todo = DBSession.query(Todo).first()
        request = testing.DummyRequest(params=params, matchdict={'id':todo.id}, post=params)
        response = update(request)
        updated_todo = DBSession.query(Todo).filter_by(id=todo.id).one()
        self.assertEqual(response['task'], params['task'])
        self.assertEqual(updated_todo.task, params['task'])
 
    def test_update_fail(self):
        """ test_update_fail """
        from todolist.controllers.todos import update
        # test query not found
        request = testing.DummyRequest(params={}, matchdict={'id':1}, post={})
        response = update(request)
        self.assertEqual(response['errors'], "No todo id: 1")
 
        # test validation
        params = {'task':"", 'priority': "low"}
        todo = DBSession.query(Todo).first()
        request = testing.DummyRequest(params=params, matchdict={'id':todo.id}, post=params)
        response = update(request)
        self.assertTrue(response['errors'])
        self.assertEqual(len(response['errors']['priority']), 1)
        self.assertEqual(response['errors']['task'], ['Please enter a number'])

It almost similar to create. One differece is a validation. I expected that priority should be a number but I put a string instead. It should display the message to let you know it’s not a number

Delete

    def test_delete_pass(self):
        """ test_delete_pass """
        from todolist.controllers.todos import delete
        todo = DBSession.query(Todo).first()
        request = testing.DummyRequest(params={},matchdict={'id':todo.id}, post={})
        response = delete(request)
        todo_count = DBSession.query(Todo).count()
        self.assertIsNot(todo, DBSession.query(Todo).first())
        self.assertEqual(response['messages'], '%s has been deleted' % todo.task)
        self.assertEqual(todo_count, 3)
 
    def test_delete_fail(self):
        """ test_delete_fail """
        from todolist.controllers.todos import delete
        # test query not found
        request = testing.DummyRequest(params={}, matchdict={'id':1}, post={})
        response = delete(request)
        self.assertEqual(response['errors'], "No todo id: 1")

These are for 2 cases too. If fails then shows a message says it has been deleted. Once deleted, the total amount should be decresed.

Then run the all codes above with nosetest, I got this…

Failed nosetest

There is a few errors. It imported the packages but not found. So I needed to write these functions which are in the next post.
Another thing in the picture above is a result of code coverage.

If that does not fix, let’s try..

You could probably use alcohol to do this as well.
Warning: do not drink Windex

To create a startup script on OSX is not hard. Just a few steps.
Firstly, make a folder in /Library/StartupItems. Then create a plist file and an executable file. That’s it.

Let’s start with open Teminal and..

$ cd /Library/StartupItems
$ sudo mkdir WhatEverNameYouWant
$ cd WhatEverNameYouWant

and then create a file named StartupParameters.plist.
You can use whatever text editor to create and add..

{
  Description     = "WhatEverNameYouWant";
  Provides        = ("ScriptNameYouWant");
  OrderPreference = "None";
}

Next, create ScriptNameYouWant and put the commands you want.
Don’t forget to start the file with shebang #!/bin/sh to execute with Bourne shell.
This can be other. It depends on what you will execute.

The file will look like..

#!/bin/sh
The commands to execute
commands
and the last command

Save and make it executable..

sudo chmod +x ScriptNameYouWant

and restart your mac. See what you fuckup your Mac lol

Let’s do a quick start.

Install a driver first. You can get one that’s written by C or Ruby. If you got gem then that’s easy.

$ sudo gem install mysql

It’s written by C

It’s quite easy to start. Require ‘mysql’ then create a connection to the database and then query. If install by gem, just require ‘rubygems’ first.

require 'rubygems'
require 'mysql'
 
db = Mysql::new("host", "user", "passwd", "db")
res = db.query("select * from mytable")
res.each do |row|
  puts "id: #{row[0]}"
end

and this is what I got

id: 1
id: 2
id: 3
id: 4
...
...

Looks simple. The index of an element in an array is sorted by the sequence of columns in the table.

Or you can get the column names instead of the indexes by each_hash

res.each_hash do |row|
  puts "id: #{row['id']}"
end

More information at the pages of the driver.

To avoid the same origin policy, JSON is a good thing to help you. The thing on the same origin policy is the browser requests from JavaScript or Flash aren’t allowed because we need to protect from script injection attacks. If this happens, the resources like http cookies will be accessed then your authentication or password in the cookies can be stolen.

So we can’t fix this but we can avoid it. The basic idea is we can request any Javascript anywhere.

<script src="http://somewhere/jsfile" type="text/javascript"></script>

Once it loaded, the browser could execute this javascript.

But if it receives json then it does nothing. We can make a returned Javascritpt looks like this.

var json_var = {'var1':'val1'};

It would be better if we can define the callback function.

myCallback({'var1':'val1'});

The server needs to provide this padding. The next thing is we have to declare this function(myCallback) to be executed once it loaded.

For instance, Graph on Facebook allows us to define the callback

https://graph.facebook.com/175459332480559/albums?callback=myCallback

by putting the ?callback=yourCallback at the end of uri

I misunderstood when I started because I saw the doc from jQuery that can put the dataType as jsonp and I thought it was just another asynchronous request which can request across the domain name on the browser.

Basically, jQuery actually provides the callback function as well like

https://graph.facebook.com/19292868552?callback=jsonp1291493679672

jsonp1291493679672 is the callback function to be defined in window like window['jsonp1291493679672'] which defines data and returns this data to the callback function we put in the ajax request. What it does is, jQuery puts ?callback=? at the end of uri and replaces a question mark(?) with jsonp1291493679672. The numbers at the end comes from now() btw.