Flask 0.10.1 rest api tutorial

enviroments

% uname -v
Darwin Kernel Version 13.4.0: Sun Aug 17 19:50:11 PDT 2014; root:xnu-2422.115.4~1/RELEASE_X86_64

% python --version
Python 3.4.2

% pip show Flask
---
Name: Flask
Version: 0.10.1
Location: /Users/okamuuu/local/python-3.4.2/lib/python3.4/site-packages
Requires: Werkzeug, Jinja2, itsdangerous

prepare

create myserver.py

import flask

app = flask.Flask(__name__)

@app.route('/')
def index():
    return "Hello, World!"

if __name__ == '__main__':
    app.run(debug=True)

create myserver_test.py

import myserver
import unittest

class MyServerTestCase(unittest.TestCase):

    def setUp(self):
        myserver.app.config['TESTING'] = True
        self.app = myserver.app.test_client()

    def test_hello(self):
        rv = self.app.get('/')
        assert b'Hello, World' in rv.data

if __name__ == '__main__':
    unittest.main()

do it.

% python myserver_test.py
.
----------------------------------------------------------------------
Ran 1 test in 0.012s

OK

edit test case and implement it.

edit myserver_test.py like this.

% git diff
diff --git a/myserver_test.py b/myserver_test.py
index 82365d9..4f8da0d 100644
--- a/myserver_test.py
+++ b/myserver_test.py
@@ -1,5 +1,6 @@
 import myserver
 import unittest
+import json
 
 class MyServerTestCase(unittest.TestCase):
 
@@ -11,5 +12,14 @@ class MyServerTestCase(unittest.TestCase):
         rv = self.app.get('/')
         assert b'Hello, World' in rv.data
 
+    def test_get_tasks(self):
+        response = self.app.get('/api/v1.0/tasks')
+        assert response.status_code == 200
+        assert response.headers['Content-Type'] == 'application/json'
+        content_body_dict = json.loads(response.data.decode())
+        tasks = content_body_dict['tasks']
+        assert tasks[0].get('id') is not None
+        assert tasks[0].get('title') is not None
+
 if __name__ == '__main__':
     unittest.main()

edit myserver.py

% git diff myserver.py
diff --git a/myserver.py b/myserver.py
index 45504a5..311c7f3 100644
--- a/myserver.py
+++ b/myserver.py
@@ -6,5 +6,17 @@ app = flask.Flask(__name__)
 def index():
     return "Hello, World!"
 
+@app.route('/api/v1.0/tasks', methods=['GET'])
+def get_tasks():
+
+    response = flask.jsonify({'tasks': [
+        {
+            'id': 1,
+            'title': 'my first todo'
+        }
+    ]})
+    response.status_code = 200
+    return response
+
 if __name__ == '__main__':
     app.run(debug=True)

test it.

% python myserver_test.py
..
----------------------------------------------------------------------
Ran 2 tests in 0.014s

OK

add test case using storage and implement it

create schema.sql

drop table if exists tasks;
create table tasks (
  id integer primary key autoincrement,
  title text not null
);

create mydb_test.py

import mydb
import unittest

class MyDbTestCase(unittest.TestCase):

    def setUp(self):
        self.db = mydb.MyDB('/tmp/test_mydb.db')
        self.db.init()

    def test_tasks(self):
        tasks = self.db.get_tasks()
        assert len(tasks) is 0
        
        self.db.add_task('title1')
        tasks = self.db.get_tasks()
        assert len(tasks) is 1

if __name__ == '__main__':
    unittest.main()

create mydb.py

import sqlite3
from contextlib import closing

class MyDB:

    def __init__(self, database):
        self.database = database

    def connect(self):
        return sqlite3.connect(self.database)

    def init(self):
        with closing(self.connect()) as db:
            with open('schema.sql', mode='r') as f:
                db.cursor().executescript(f.read())
            db.commit()

    def get_tasks(self):
      db = self.connect()
      cur = db.execute('select * from tasks order by id desc')
      tasks = [dict(id=row[0], title=row[1]) for row in cur.fetchall()]
      db.close();
      return tasks

    def add_task(self, title):
      db = self.connect()
      db.execute('insert into tasks (title) values (?)', [title])
      db.commit()
      db.close()

test it.

% python mydb_test.py
.
----------------------------------------------------------------------
Ran 1 test in 0.005s

OK

edit myserver.py

diff --git a/myserver.py b/myserver.py
index 311c7f3..6e9c418 100644
--- a/myserver.py
+++ b/myserver.py
@@ -1,7 +1,19 @@
 import flask
+import mydb
+import os
 
 app = flask.Flask(__name__)

+if os.getenv('TEST') == 'True':
+    app.config['DATABASE'] = '/tmp/test_mydb.db'
+else:
+    app.config['DATABASE'] = '/tmp/mydb.db'
+
+db = mydb.MyDB(app.config['DATABASE'])
+db.init()
+
 @app.route('/')
 def index():
     return "Hello, World!"
@@ -9,13 +21,24 @@ def index():
 @app.route('/api/v1.0/tasks', methods=['GET'])
 def get_tasks():
 
-    response = flask.jsonify({'tasks': [
-        {
-            'id': 1,
-            'title': 'my first todo'
-        }
-    ]})
+    response = flask.jsonify({'tasks': db.get_tasks()})
     response.status_code = 200
+    
+    return response
+
+@app.route('/api/v1.0/tasks', methods=['POST'])
+def add_task():
+
+    request = flask.request
+    
+    if not (request.json and 'title' in request.json):
+        flask.abort(400)
+
+    db.add_task(request.json['title'])
+    
+    response = flask.jsonify({'status':'ok'})
+    response.status_code = 201
+    
     return response
 
 if __name__ == '__main__':

edit myserver_test.py

diff --git a/myserver_test.py b/myserver_test.py
index 4f8da0d..b1dac21 100644
--- a/myserver_test.py
+++ b/myserver_test.py
@@ -1,3 +1,6 @@
+import os
+os.environ["TEST"] = 'True'
+
 import myserver
 import unittest
 import json
@@ -5,19 +8,26 @@ import json
 class MyServerTestCase(unittest.TestCase):
 
     def setUp(self):
-        myserver.app.config['TESTING'] = True
         self.app = myserver.app.test_client()
 
     def test_hello(self):
         rv = self.app.get('/')
         assert b'Hello, World' in rv.data
 
-    def test_get_tasks(self):
+    def test_task_api(self):
+        response = self.app.get('/api/v1.0/tasks')
+        rv = self.app.post(
+            '/api/v1.0/tasks', 
+            content_type='application/json', 
+            data=json.dumps({"title":"Hello"})
+        )
+
         response = self.app.get('/api/v1.0/tasks')
         assert response.status_code == 200
         assert response.headers['Content-Type'] == 'application/json'
         content_body_dict = json.loads(response.data.decode())
         tasks = content_body_dict['tasks']
+        assert len(tasks) is 1
         assert tasks[0].get('id') is not None
         assert tasks[0].get('title') is not None

test it

% python myserver_test.py
..
----------------------------------------------------------------------
Ran 2 tests in 0.017s

OK

check

run it

% python myserver.py

and check it

% curl http://localhost:5000/api/v1.0/tasks
{
  "tasks": []
}

% curl -i -X POST -H "Content-Type: application/json" -d '{"title":"title1"}' http://localhost:5000/api/v1.0/tasks
HTTP/1.0 201 CREATED
Content-Type: application/json
Content-Length: 20
Server: Werkzeug/0.9.6 Python/3.4.2
Date: Fri, 28 Nov 2014 04:27:19 GMT

{
  "status": "ok"
}

% curl http://localhost:5000/api/v1.0/tasks
{
  "tasks": [
    {
      "id": 1,
      "title": "title1"
    }
  ]
}

enjoy:)