1. 程式人生 > 其它 >Flask 設計 RESTful API

Flask 設計 RESTful API

使用 Python 和 Flask 設計 RESTful API

近些年來 REST (REpresentational State Transfer) 已經變成了 web services 和 web APIs 的標配。

在本文中我將向你展示如何簡單地使用 Python 和 Flask 框架來建立一個 RESTful 的 web service。

什麼是 REST?

六條設計規範定義了一個 REST 系統的特點:

  • 客戶端-伺服器: 客戶端和伺服器之間隔離,伺服器提供服務,客戶端進行消費。
  • 無狀態: 從客戶端到伺服器的每個請求都必須包含理解請求所必需的資訊。換句話說, 伺服器不會儲存客戶端上一次請求的資訊用來給下一次使用。
  • 可快取: 伺服器必須明示客戶端請求能否快取。
  • 分層系統: 客戶端和伺服器之間的通訊應該以一種標準的方式,就是中間層代替伺服器做出響應的時候,客戶端不需要做任何變動。
  • 統一的介面: 伺服器和客戶端的通訊方法必須是統一的。
  • 按需編碼: 伺服器可以提供可執行程式碼或指令碼,為客戶端在它們的環境中執行。這個約束是唯一一個是可選的。

什麼是一個 RESTful 的 web service?

REST 架構的最初目的是適應全球資訊網的 HTTP 協議。

RESTful web services 概念的核心就是“資源”。 資源可以用URI來表示。客戶端使用 HTTP 協議定義的方法來發送請求到這些 URIs,當然可能會導致這些被訪問的”資源“狀態的改變。

HTTP 標準的方法有如下:

==========  =====================  ==================================
HTTP 方法   行為                   示例
==========  =====================  ==================================
GET         獲取資源的資訊         http://example.com/api/orders
GET         獲取某個特定資源的資訊 http://example.com/api/orders/123
POST        建立新資源             http://example.com/api/orders
PUT         更新資源               http://example.com/api/orders/123
DELETE      刪除資源               http://example.com/api/orders/123
==========  ====================== ==================================

REST 設計不需要特定的資料格式。在請求中資料可以以JSON形式, 或者有時候作為 url 中查詢引數項。

設計一個簡單的 web service

堅持 REST 的準則設計一個 web service 或者 API 的任務就變成一個標識資源被展示出來以及它們是怎樣受不同的請求方法影響的練習。

比如說,我們要編寫一個待辦事項應用程式而且我們想要為它設計一個 web service。要做的第一件事情就是決定用什麼樣的根 URL 來訪問該服務。例如,我們可以通過這個來訪問:

http://[hostname]/todo/api/v1.0/

在這裡我已經決定在 URL 中包含應用的名稱以及 API 的版本號。在 URL 中包含應用名稱有助於提供一個名稱空間以便區分同一系統上的其它服務。在 URL 中包含版本號能夠幫助以後的更新,如果新版本中存在新的和潛在不相容的功能,可以不影響依賴於較舊的功能的應用程式。

下一步驟就是選擇將由該服務暴露(展示)的資源。這是一個十分簡單地應用,我們只有任務,因此在我們待辦事項中唯一的資源就是任務。

我們的任務資源將要使用 HTTP 方法如下:

==========  ===============================================  =============================
HTTP 方法   URL                                              動作
==========  ===============================================  ==============================
GET         http://[hostname]/todo/api/v1.0/tasks            檢索任務列表
GET         http://[hostname]/todo/api/v1.0/tasks/[task_id]  檢索某個任務
POST        http://[hostname]/todo/api/v1.0/tasks            建立新任務
PUT         http://[hostname]/todo/api/v1.0/tasks/[task_id]  更新任務
DELETE      http://[hostname]/todo/api/v1.0/tasks/[task_id]  刪除任務
==========  ================================================ =============================

我們定義的任務有如下一些屬性:

  • id: 任務的唯一識別符號。數字型別。
  • title: 簡短的任務描述。字串型別。
  • description: 具體的任務描述。文字型別。
  • done: 任務完成的狀態。布林值。

目前為止關於我們的 web service 的設計基本完成。剩下的事情就是實現它!

Flask 框架的簡介

如果你讀過Flask Mega-Tutorial 系列,就會知道 Flask 是一個簡單卻十分強大的 Python web 框架。

在我們深入研究 web services 的細節之前,讓我們回顧一下一個普通的 Flask Web 應用程式的結構。

我會首先假設你知道 Python 在你的平臺上工作的基本知識。 我將講解的例子是工作在一個類 Unix 作業系統。簡而言之,這意味著它們能工作在 Linux,Mac OS X 和 Windows(如果你使用Cygwin)。 如果你使用 Windows 上原生的 Python 版本的話,命令會有所不同。

讓我們開始在一個虛擬環境上安裝 Flask。如果你的系統上沒有 virtualenv,你可以從https://pypi.python.org/pypi/virtualenv上下載:

$ mkdir todo-api
$ cd todo-api
$ virtualenv flask
New python executable in flask/bin/python
Installing setuptools............................done.
Installing pip...................done.
$ flask/bin/pip install flask

既然已經安裝了 Flask,現在開始建立一個簡單地網頁應用,我們把它放在一個叫 app.py 的檔案中:

#!flask/bin/python
from flask import Flask

app = Flask(__name__)

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

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

為了執行這個程式我們必須執行 app.py:

$ chmod a+x app.py
$ ./app.py
 * Running on http://127.0.0.1:5000/
 * Restarting with reloader

現在你可以啟動你的網頁瀏覽器,輸入http://localhost:5000看看這個小應用程式的效果。

簡單吧?現在我們將這個應用程式轉換成我們的 RESTful service!

使用 Python 和 Flask 實現 RESTful services

使用 Flask 構建 web services 是十分簡單地,比我在Mega-Tutorial中構建的完整的服務端的應用程式要簡單地多。

在 Flask 中有許多擴充套件來幫助我們構建 RESTful services,但是在我看來這個任務十分簡單,沒有必要使用 Flask 擴充套件。

我們 web service 的客戶端需要新增、刪除以及修改任務的服務,因此顯然我們需要一種方式來儲存任務。最直接的方式就是建立一個小型的資料庫,但是資料庫並不是本文的主體。學習在 Flask 中使用合適的資料庫,我強烈建議閱讀Mega-Tutorial

這裡我們直接把任務列表儲存在記憶體中,因此這些任務列表只會在 web 伺服器執行中工作,在結束的時候就失效。 這種方式只是適用我們自己開發的 web 伺服器,不適用於生產環境的 web 伺服器, 這種情況一個合適的資料庫的搭建是必須的。

我們現在來實現 web service 的第一個入口:

#!flask/bin/python
from flask import Flask, jsonify

app = Flask(__name__)

tasks = [
    {
        'id': 1,
        'title': u'Buy groceries',
        'description': u'Milk, Cheese, Pizza, Fruit, Tylenol',
        'done': False
    },
    {
        'id': 2,
        'title': u'Learn Python',
        'description': u'Need to find a good Python tutorial on the web',
        'done': False
    }
]

@app.route('/todo/api/v1.0/tasks', methods=['GET'])
def get_tasks():
    return jsonify({'tasks': tasks})

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

正如你所見,沒有多大的變化。我們建立一個任務的記憶體資料庫,這裡無非就是一個字典和陣列。陣列中的每一個元素都具有上述定義的任務的屬性。

取代了首頁,我們現在擁有一個 get_tasks 的函式,訪問的 URI 為 /todo/api/v1.0/tasks,並且只允許 GET 的 HTTP 方法。

這個函式的響應不是文字,我們使用 JSON 資料格式來響應,Flask 的 jsonify 函式從我們的資料結構中生成。

使用網頁瀏覽器來測試我們的 web service 不是一個最好的注意,因為網頁瀏覽器上不能輕易地模擬所有的 HTTP 請求的方法。相反,我們會使用 curl。如果你還沒有安裝 curl 的話,請立即安裝它。

通過執行 app.py,啟動 web service。接著開啟一個新的控制檯視窗,執行以下命令:

$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 294
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 04:53:53 GMT

{
  "tasks": [
    {
      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",
      "done": false,
      "id": 1,
      "title": "Buy groceries"
    },
    {
      "description": "Need to find a good Python tutorial on the web",
      "done": false,
      "id": 2,
      "title": "Learn Python"
    }
  ]
}

我們已經成功地呼叫我們的 RESTful service 的一個函式!

現在我們開始編寫 GET 方法請求我們的任務資源的第二個版本。這是一個用來返回單獨一個任務的函式:

from flask import abort

@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id):
    task = filter(lambda t: t['id'] == task_id, tasks)
    if len(task) == 0:
        abort(404)
    return jsonify({'task': task[0]})

第二個函式有些意思。這裡我們得到了 URL 中任務的 id,接著 Flask 把它轉換成 函式中的 task_id 的引數。

我們用這個引數來搜尋我們的任務陣列。如果我們的資料庫中不存在搜尋的 id,我們將會返回一個類似 404 的錯誤,根據 HTTP 規範的意思是 “資源未找到”。

如果我們找到相應的任務,那麼我們只需將它用 jsonify 打包成 JSON 格式並將其傳送作為響應,就像我們以前那樣處理整個任務集合。

呼叫 curl 請求的結果如下:

$ curl -i http://localhost:5000/todo/api/v1.0/tasks/2
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 151
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 05:21:50 GMT

{
  "task": {
    "description": "Need to find a good Python tutorial on the web",
    "done": false,
    "id": 2,
    "title": "Learn Python"
  }
}
$ curl -i http://localhost:5000/todo/api/v1.0/tasks/3
HTTP/1.0 404 NOT FOUND
Content-Type: text/html
Content-Length: 238
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 05:21:52 GMT

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
<title>404 Not Found</title>
<h1>Not Found</h1>
<p>The requested URL was not found on the server.</p><p>If you     entered the URL manually please check your spelling and try again.</p>

當我們請求 id #2 的資源時候,我們獲取到了,但是當我們請求 #3 的時候返回了 404 錯誤。有關錯誤奇怪的是返回的是 HTML 資訊而不是 JSON,這是因為 Flask 按照預設方式生成 404 響應。由於這是一個 Web service 客戶端希望我們總是以 JSON 格式迴應,所以我們需要改善我們的 404 錯誤處理程式:

from flask import make_response

@app.errorhandler(404)
def not_found(error):
    return make_response(jsonify({'error': 'Not found'}), 404)

我們會得到一個友好的錯誤提示:

$ curl -i http://localhost:5000/todo/api/v1.0/tasks/3
HTTP/1.0 404 NOT FOUND
Content-Type: application/json
Content-Length: 26
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 05:36:54 GMT

{
  "error": "Not found"
}

接下來就是 POST 方法,我們用來在我們的任務資料庫中插入一個新的任務:

from flask import request

@app.route('/todo/api/v1.0/tasks', methods=['POST'])
def create_task():
    if not request.json or not 'title' in request.json:
        abort(400)
    task = {
        'id': tasks[-1]['id'] + 1,
        'title': request.json['title'],
        'description': request.json.get('description', ""),
        'done': False
    }
    tasks.append(task)
    return jsonify({'task': task}), 201

新增一個新的任務也是相當容易地。只有當請求以 JSON 格式形式,request.json 才會有請求的資料。如果沒有資料,或者存在資料但是缺少 title 項,我們將會返回 400,這是表示請求無效。

接著我們會建立一個新的任務字典,使用最後一個任務的 id + 1 作為該任務的 id。我們允許 description 欄位缺失,並且假設 done 欄位設定成 False。

我們把新的任務新增到我們的任務陣列中,並且把新新增的任務和狀態 201 響應給客戶端。

使用如下的 curl 命令來測試這個新的函式:

$ curl -i -H "Content-Type: application/json" -X POST -d '{"title":"Read a book"}' http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.0 201 Created
Content-Type: application/json
Content-Length: 104
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 05:56:21 GMT

{
  "task": {
    "description": "",
    "done": false,
    "id": 3,
    "title": "Read a book"
  }
}

注意:如果你在 Windows 上並且執行 Cygwin 版本的 curl,上面的命令不會有任何問題。然而,如果你使用原生的 curl,命令會有些不同:

curl -i -H "Content-Type: application/json" -X POST -d "{"""title""":"""Read a book"""}" http://localhost:5000/todo/api/v1.0/tasks

當然在完成這個請求後,我們可以得到任務的更新列表:

$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 423
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 05:57:44 GMT

{
  "tasks": [
    {
      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",
      "done": false,
      "id": 1,
      "title": "Buy groceries"
    },
    {
      "description": "Need to find a good Python tutorial on the web",
      "done": false,
      "id": 2,
      "title": "Learn Python"
    },
    {
      "description": "",
      "done": false,
      "id": 3,
      "title": "Read a book"
    }
  ]
}

剩下的兩個函式如下所示:

@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
    task = filter(lambda t: t['id'] == task_id, tasks)
    if len(task) == 0:
        abort(404)
    if not request.json:
        abort(400)
    if 'title' in request.json and type(request.json['title']) != unicode:
        abort(400)
    if 'description' in request.json and type(request.json['description']) is not unicode:
        abort(400)
    if 'done' in request.json and type(request.json['done']) is not bool:
        abort(400)
    task[0]['title'] = request.json.get('title', task[0]['title'])
    task[0]['description'] = request.json.get('description', task[0]['description'])
    task[0]['done'] = request.json.get('done', task[0]['done'])
    return jsonify({'task': task[0]})

@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    task = filter(lambda t: t['id'] == task_id, tasks)
    if len(task) == 0:
        abort(404)
    tasks.remove(task[0])
    return jsonify({'result': True})

delete_task 函式沒有什麼特別的。對於 update_task 函式,我們需要嚴格地檢查輸入的引數以防止可能的問題。我們需要確保在我們把它更新到資料庫之前,任何客戶端提供我們的是預期的格式。

更新任務 #2 的函式呼叫如下所示:

$ curl -i -H "Content-Type: application/json" -X PUT -d '{"done":true}' http://localhost:5000/todo/api/v1.0/tasks/2
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 170
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 07:10:16 GMT

{
  "task": [
    {
      "description": "Need to find a good Python tutorial on the web",
      "done": true,
      "id": 2,
      "title": "Learn Python"
    }
  ]
}

優化 web service 介面

目前 API 的設計的問題就是迫使客戶端在任務標識返回後去構造 URIs。這對於伺服器是十分簡單的,但是間接地迫使客戶端知道這些 URIs 是如何構造的,這將會阻礙我們以後變更這些 URIs。

不直接返回任務的 ids,我們直接返回控制這些任務的完整的 URI,以便客戶端可以隨時使用這些 URIs。為此,我們可以寫一個小的輔助函式生成一個 “公共” 版本任務傳送到客戶端:

from flask import url_for

def make_public_task(task):
    new_task = {}
    for field in task:
        if field == 'id':
            new_task['uri'] = url_for('get_task', task_id=task['id'], _external=True)
        else:
            new_task[field] = task[field]
    return new_task

這裡所有做的事情就是從我們資料庫中取出任務並且建立一個新的任務,這個任務的 id 欄位被替換成通過 Flask 的 url_for 生成的 uri 欄位。

當我們返回所有的任務列表的時候,在傳送到客戶端之前通過這個函式進行處理:

@app.route('/todo/api/v1.0/tasks', methods=['GET'])
def get_tasks():
    return jsonify({'tasks': map(make_public_task, tasks)})

這裡就是客戶端獲取任務列表的時候得到的資料:

$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 406
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 18:16:28 GMT

{
  "tasks": [
    {
      "title": "Buy groceries",
      "done": false,
      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",
      "uri": "http://localhost:5000/todo/api/v1.0/tasks/1"
    },
    {
      "title": "Learn Python",
      "done": false,
      "description": "Need to find a good Python tutorial on the web",
      "uri": "http://localhost:5000/todo/api/v1.0/tasks/2"
    }
  ]
}

我們將會把上述的方式應用到其它所有的函式上以確保客戶端一直看到 URIs 而不是 ids。

加強 RESTful web service 的安全性

我們已經完成了我們 web service 的大部分功能,但是仍然有一個問題。我們的 web service 對任何人都是公開的,這並不是一個好主意。

我們有一個可以管理我們的待辦事項完整的 web service,但在當前狀態下的 web service 是開放給所有的客戶端。 如果一個陌生人弄清我們的 API 是如何工作的,他或她可以編寫一個客戶端訪問我們的 web service 並且毀壞我們的資料。

大部分初級的教程會忽略這個問題並且到此為止。在我看來這是一個很嚴重的問題,我必須指出。

確保我們的 web service 安全服務的最簡單的方法是要求客戶端提供一個使用者名稱和密碼。在常規的 web 應用程式會提供一個登入的表單用來認證,並且伺服器會建立一個會話為登入的使用者以後的操作使用,會話的 id 以 cookie 形式儲存在客戶端瀏覽器中。然而 REST 的規則之一就是 “無狀態”, 因此我們必須要求客戶端在每一次請求中提供認證的資訊。

我們一直試著儘可能地堅持 HTTP 標準協議。既然我們需要實現認證我們需要在 HTTP 上下文中去完成,HTTP 協議提供了兩種認證機制:Basic 和 Digest

有一個小的 Flask 擴充套件能夠幫助我們,我們可以先安裝 Flask-HTTPAuth:

$ flask/bin/pip install flask-httpauth

比方說,我們希望我們的 web service 只讓訪問使用者名稱 miguel 和密碼 python 的客戶端訪問。 我們可以設定一個基本的 HTTP 驗證如下:

from flask.ext.httpauth import HTTPBasicAuth
auth = HTTPBasicAuth()

@auth.get_password
def get_password(username):
    if username == 'miguel':
        return 'python'
    return None

@auth.error_handler
def unauthorized():
    return make_response(jsonify({'error': 'Unauthorized access'}), 401)

get_password 函式是一個回撥函式,Flask-HTTPAuth 使用它來獲取給定使用者的密碼。在一個更復雜的系統中,這個函式是需要檢查一個使用者資料庫,但是在我們的例子中只有單一的使用者因此沒有必要。

error_handler 回撥函式是用於給客戶端傳送未授權錯誤程式碼。像我們處理其它的錯誤程式碼,這裡我們定製一個包含 JSON 資料格式而不是 HTML 的響應。

隨著認證系統的建立,所剩下的就是把需要認證的函式新增 @auth.login_required 裝飾器。例如:

@app.route('/todo/api/v1.0/tasks', methods=['GET'])
@auth.login_required
def get_tasks():
    return jsonify({'tasks': tasks})

如果現在要嘗試使用 curl 呼叫這個函式我們會得到:

$ curl -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.0 401 UNAUTHORIZED
Content-Type: application/json
Content-Length: 36
WWW-Authenticate: Basic realm="Authentication Required"
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 06:41:14 GMT

{
  "error": "Unauthorized access"
}

為了能夠呼叫這個函式我們必須傳送我們的認證憑據:

$ curl -u miguel:python -i http://localhost:5000/todo/api/v1.0/tasks
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 316
Server: Werkzeug/0.8.3 Python/2.7.3
Date: Mon, 20 May 2013 06:46:45 GMT

{
  "tasks": [
    {
      "title": "Buy groceries",
      "done": false,
      "description": "Milk, Cheese, Pizza, Fruit, Tylenol",
      "uri": "http://localhost:5000/todo/api/v1.0/tasks/1"
    },
    {
      "title": "Learn Python",
      "done": false,
      "description": "Need to find a good Python tutorial on the web",
      "uri": "http://localhost:5000/todo/api/v1.0/tasks/2"
    }
  ]
}

認證擴充套件給予我們很大的自由選擇哪些函式需要保護,哪些函式需要公開。

為了確保登入資訊的安全應該使用 HTTP 安全伺服器(例如:https://...),這樣客戶端和伺服器之間的通訊都是加密的,以防止傳輸過程中第三方看到認證的憑據。

讓人不舒服的是當請求收到一個 401 的錯誤,網頁瀏覽都會跳出一個醜陋的登入框,即使請求是在後臺發生的。因此如果我們要實現一個完美的 web 伺服器的話,我們就需要禁止跳轉到瀏覽器顯示身份驗證對話方塊,讓我們的客戶端應用程式自己處理登入。

一個簡單的方式就是不返回 401 錯誤。403 錯誤是一個令人青睞的替代,403 錯誤表示 “禁止” 的錯誤:

@auth.error_handler
def unauthorized():
    return make_response(jsonify({'error': 'Unauthorized access'}), 403)

可能的改進

我們編寫的小型的 web service 還可以在不少的方面進行改進。

對於初學者來說,一個真正的 web service 需要一個真實的資料庫進行支撐。我們現在使用的記憶體資料結構會有很多限制不應該被用於真正的應用。

另外一個可以提高的領域就是處理多使用者。如果系統支援多使用者的話,不同的客戶端可以傳送不同的認證憑證獲取相應使用者的任務列表。在這樣一個系統中的話,我們需要第二個資源就是使用者。在使用者資源上的 POST 的請求代表註冊換一個新使用者。一個 GET 請求表示客戶端獲取一個使用者的資訊。一個 PUT 請求表示更新使用者資訊,比如可能是更新郵箱地址。一個 DELETE 請求表示刪除使用者賬號。

GET 檢索任務列表請求可以在幾個方面進行擴充套件。首先可以攜帶一個可選的頁的引數,以便客戶端請求任務的一部分。另外,這種擴充套件更加有用:允許按照一定的標準篩選。比如,使用者只想要看到完成的任務,或者只想看到任務的標題以 A 字母開頭。所有的這些都可以作為 URL 的一個引數項。

http://www.pythondoc.com/flask-restful/first.html