筆記

RESTful架構(gòu)風(fēng)格概述

RESTful架構(gòu)風(fēng)格

RESTful架構(gòu)風(fēng)格最初由Roy T. Fielding（HTTP/1.1協(xié)議專家組負(fù)責(zé)人）在其2000年的博士學(xué)位論文中提出漆羔。HTTP就是該架構(gòu)風(fēng)格的一個(gè)典型應(yīng)用俐末。

REST即Representational State Transfer的縮寫，可譯為"表現(xiàn)層狀態(tài)轉(zhuǎn)化”眉菱。REST最大的幾個(gè)特點(diǎn)為：資源、統(tǒng)一接口蛇摸、URI和無狀態(tài)摔吏。

RESTful架構(gòu)風(fēng)格的特點(diǎn)

資源

所謂"資源"愉镰，就是網(wǎng)絡(luò)上的一個(gè)實(shí)體骨宠，或者說是網(wǎng)絡(luò)上的一個(gè)具體信息浮定。它可以是一段文本相满、一張圖片、一首歌曲桦卒、一種服務(wù)雳灵，總之就是一個(gè)具體的實(shí)在。資源總要通過某種載體反應(yīng)其內(nèi)容闸盔，文本可以用txt格式表現(xiàn)，也可以用HTML格式琳省、XML格式表現(xiàn)迎吵，甚至可以采用二進(jìn)制格式；圖片可以用JPG格式表現(xiàn)针贬，也可以用PNG格式表現(xiàn)击费；JSON是現(xiàn)在最常用的資源表示格式。

統(tǒng)一接口

RESTful架構(gòu)風(fēng)格規(guī)定桦他，數(shù)據(jù)的元操作蔫巩，即CRUD(create, read, update和delete,即數(shù)據(jù)的增刪查改)操作，分別對(duì)應(yīng)于HTTP方法：GET用來獲取資源快压，POST用來新建資源（也可以用于更新資源）圆仔，PUT用來更新資源，DELETE用來刪除資源蔫劣，這樣就統(tǒng)一了數(shù)據(jù)操作的接口坪郭，僅通過HTTP方法，就可以完成對(duì)數(shù)據(jù)的所有增刪查改工作。

• GET（SELECT）：從服務(wù)器取出資源（一項(xiàng)或多項(xiàng)）。
• POST（CREATE）：在服務(wù)器新建一個(gè)資源日月。
• PUT（UPDATE）：在服務(wù)器更新資源（客戶端提供完整資源數(shù)據(jù)）幽告。
• DELETE（DELETE）：從服務(wù)器刪除資源。

URI

可以用一個(gè)URI（統(tǒng)一資源定位符）指向資源鹊奖，即每個(gè)URI都對(duì)應(yīng)一個(gè)特定的資源。要獲取這個(gè)資源，訪問它的URI就可以液走，因此URI就成了每一個(gè)資源的地址或識(shí)別符。

一般的贾陷，每個(gè)資源至少有一個(gè)URI與之對(duì)應(yīng)育灸，最典型的URI即URL。

無狀態(tài)

所謂無狀態(tài)的昵宇，即所有的資源磅崭，都可以通過URI定位，而且這個(gè)定位與其他資源無關(guān)瓦哎，也不會(huì)因?yàn)槠渌Y源的變化而改變砸喻。有狀態(tài)和無狀態(tài)的區(qū)別柔逼，舉個(gè)簡單的例子說明一下。如查詢員工的工資割岛，如果查詢工資是需要登錄系統(tǒng)愉适，進(jìn)入查詢工資的頁面，執(zhí)行相關(guān)操作后癣漆，獲取工資的多少维咸，則這種情況是有狀態(tài)的，因?yàn)椴樵児べY的每一步操作都依賴于前一步操作惠爽，只要前置操作不成功癌蓖，后續(xù)操作就無法執(zhí)行；如果輸入一個(gè)url即可得到指定員工的工資婚肆，則這種情況是無狀態(tài)的租副，因?yàn)楂@取工資不依賴于其他資源或狀態(tài)，且這種情況下较性，員工工資是一個(gè)資源用僧，由一個(gè)url與之對(duì)應(yīng)，可以通過HTTP中的GET方法得到資源赞咙，這是典型的RESTful風(fēng)格责循。

ROA

ROA即Resource Oriented Architecture，RESTful 架構(gòu)風(fēng)格的服務(wù)是圍繞資源展開的攀操，是典型的ROA架構(gòu)沼死。RESTful 架構(gòu)風(fēng)格的服務(wù)通常被稱之為ROA架構(gòu)。

RESTful風(fēng)格的服務(wù)崔赌，由于可以直接以json或xml為載體承載數(shù)據(jù)意蛀，以HTTP方法為統(tǒng)一接口完成數(shù)據(jù)操作，客戶端的開發(fā)不依賴于服務(wù)實(shí)現(xiàn)的技術(shù)健芭，移動(dòng)終端也可以輕松使用服務(wù)县钥，這也加劇了REST取代RPC成為web service的主導(dǎo)。

認(rèn)證機(jī)制

認(rèn)證機(jī)制基本上是通用的慈迈，常用的認(rèn)證機(jī)制包括 session auth(即通過用戶名密碼登錄)若贮，basic auth，token auth和OAuth痒留，服務(wù)開發(fā)中常用的認(rèn)證機(jī)制為后三者谴麦。

Basic Auth

Basic Auth是配合RESTful API 使用的最簡單的認(rèn)證方式，只需提供用戶名密碼即可伸头，但由于有把用戶名密碼暴露給第三方客戶端的風(fēng)險(xiǎn)匾效，在生產(chǎn)環(huán)境下被使用的越來越少。

Token Auth

Token Auth并不常用恤磷，它與Basic Auth的區(qū)別是面哼，不將用戶名和密碼發(fā)送給服務(wù)器做用戶認(rèn)證野宜，而是向服務(wù)器發(fā)送一個(gè)事先在服務(wù)器端生成的token來做認(rèn)證。因此Token Auth要求服務(wù)器端要具備一套完整的Token創(chuàng)建和管理機(jī)制魔策，該機(jī)制的實(shí)現(xiàn)會(huì)增加大量且非必須的服務(wù)器端開發(fā)工作匈子，也不見得這套機(jī)制足夠安全和通用，因此Token Auth用的并不多闯袒。

OAuth

OAuth（開放授權(quán)）是一個(gè)開放的授權(quán)標(biāo)準(zhǔn)虎敦，允許用戶讓第三方應(yīng)用訪問該用戶在某一web服務(wù)上存儲(chǔ)的私密的資源（如照片，視頻政敢，聯(lián)系人列表）其徙，而無需將用戶名和密碼提供給第三方應(yīng)用。

OAuth允許用戶提供一個(gè)令牌堕仔，而不是用戶名和密碼來訪問他們存放在特定服務(wù)提供者的數(shù)據(jù)。每一個(gè)令牌授權(quán)一個(gè)特定的第三方系統(tǒng)（例如晌区，視頻編輯網(wǎng)站)在特定的時(shí)段（例如摩骨，接下來的2小時(shí)內(nèi)）內(nèi)訪問特定的資源（例如僅僅是某一相冊中的視頻）。這樣朗若，OAuth讓用戶可以授權(quán)第三方網(wǎng)站訪問他們存儲(chǔ)在另外服務(wù)提供者的某些特定信息恼五，而非所有內(nèi)容。

正是由于OAUTH的嚴(yán)謹(jǐn)性和安全性哭懈，現(xiàn)在OAUTH已成為RESTful架構(gòu)風(fēng)格中最常用的認(rèn)證機(jī)制灾馒，和RESTful架構(gòu)風(fēng)格一起，成為企業(yè)級(jí)服務(wù)的標(biāo)配遣总。

RESTful API 編寫指南

Request和Response

RESTful API的開發(fā)和使用睬罗，無非是客戶端向服務(wù)器發(fā)請(qǐng)求（request），以及服務(wù)器對(duì)客戶端請(qǐng)求的響應(yīng)（response）旭斥。

響應(yīng)這些request時(shí)容达，常用的Response要包含的數(shù)據(jù)和狀態(tài)碼（status code）

• 當(dāng)GET和PUT請(qǐng)求成功時(shí)，要返回對(duì)應(yīng)的數(shù)據(jù)垂券，及狀態(tài)碼200花盐，即SUCCESS
• 當(dāng)POST創(chuàng)建數(shù)據(jù)成功時(shí)，要返回創(chuàng)建的數(shù)據(jù)菇爪，及狀態(tài)碼201算芯，即CREATED
• 當(dāng)DELETE刪除數(shù)據(jù)成功時(shí)，不返回?cái)?shù)據(jù)凳宙，狀態(tài)碼要返回204熙揍，即NO CONTENT
• 當(dāng)GET 不到數(shù)據(jù)時(shí)，狀態(tài)碼要返回404氏涩，即NOT FOUND
• 任何時(shí)候诈嘿，如果請(qǐng)求有問題堪旧，如校驗(yàn)請(qǐng)求數(shù)據(jù)時(shí)發(fā)現(xiàn)錯(cuò)誤，要返回狀態(tài)碼 400奖亚，即BAD REQUEST
• 當(dāng)API 請(qǐng)求需要用戶認(rèn)證時(shí)淳梦，如果request中的認(rèn)證信息不正確，要返回狀態(tài)碼 401昔字，即NOT AUTHORIZED
• 當(dāng)API 請(qǐng)求需要驗(yàn)證用戶權(quán)限時(shí)爆袍，如果當(dāng)前用戶無相應(yīng)權(quán)限，要返回狀態(tài)碼 403作郭，即FORBIDDEN

最后陨囊，關(guān)于Request 和 Response，不要忽略了http header中的Content-Type夹攒。以json為例蜘醋，如果API要求客戶端發(fā)送request時(shí)要傳入json數(shù)據(jù)，則服務(wù)器端僅做好json數(shù)據(jù)的獲取和解析即可咏尝，但如果服務(wù)端支持多種類型數(shù)據(jù)的傳入压语，如同時(shí)支持json和form-data，則要根據(jù)客戶端發(fā)送請(qǐng)求時(shí)header中的Content-Type编检，對(duì)不同類型是數(shù)據(jù)分別實(shí)現(xiàn)獲取和解析胎食；如果API響應(yīng)客戶端請(qǐng)求后，需要返回json數(shù)據(jù)允懂，需要在header中添加Content-Type=application/json厕怜。

Serialization和Deserialization

Serialization和Deserialization即序列化和反序列化。RESTful API以規(guī)范統(tǒng)一的格式作為數(shù)據(jù)的載體蕾总，常用的格式為json或xml粥航，以json格式為例，當(dāng)客戶端向服務(wù)器發(fā)請(qǐng)求時(shí)生百，或者服務(wù)器相應(yīng)客戶端的請(qǐng)求躁锡，向客戶端返回?cái)?shù)據(jù)時(shí)，都是傳輸json格式的文本置侍，而在服務(wù)器內(nèi)部映之，數(shù)據(jù)處理時(shí)基本不用json格式的字符串，而是native類型的數(shù)據(jù)蜡坊，最典型的如類的實(shí)例杠输，即對(duì)象（object），json僅為服務(wù)器和客戶端通信時(shí)秕衙，在網(wǎng)絡(luò)上傳輸?shù)臄?shù)據(jù)的格式蠢甲，服務(wù)器和客戶端內(nèi)部，均存在將json轉(zhuǎn)為native類型數(shù)據(jù)和將native類型數(shù)據(jù)轉(zhuǎn)為json的需求据忘，其中鹦牛，將native類型數(shù)據(jù)轉(zhuǎn)為json即為序列化搞糕，將json轉(zhuǎn)為native類型數(shù)據(jù)即為反序列化。

我們在開發(fā)RESTful API時(shí)曼追，沒必要制造重復(fù)的輪子窍仰，選一個(gè)好用的庫即可，如python中的marshmallow礼殊。

Validation

Validation即數(shù)據(jù)校驗(yàn)驹吮，是開發(fā)健壯RESTful API中另一個(gè)重要的一環(huán)。仍以json為例晶伦，當(dāng)客戶端向服務(wù)器發(fā)出post或put請(qǐng)求時(shí)碟狞，通常會(huì)同時(shí)給服務(wù)器發(fā)送json格式的相關(guān)數(shù)據(jù)，服務(wù)器在做數(shù)據(jù)處理之前婚陪，先做數(shù)據(jù)校驗(yàn)族沃，是最合理和安全的前后端交互。如果客戶端發(fā)送的數(shù)據(jù)不正確或不合理泌参，服務(wù)器端經(jīng)過校驗(yàn)后直接向客戶端返回400錯(cuò)誤及相應(yīng)的數(shù)據(jù)錯(cuò)誤信息即可脆淹。常見的數(shù)據(jù)校驗(yàn)包括：

• 數(shù)據(jù)類型校驗(yàn)，如字段類型如果是int及舍，那么給字段賦字符串的值則報(bào)錯(cuò)
• 數(shù)據(jù)格式校驗(yàn)未辆，如郵箱或密碼窟绷，其賦值必須滿足相應(yīng)的正則表達(dá)式锯玛，才是正確的輸入數(shù)據(jù)
• 數(shù)據(jù)邏輯校驗(yàn)，如數(shù)據(jù)包含出生日期和年齡兩個(gè)字段兼蜈，如果這兩個(gè)字段的數(shù)據(jù)不一致攘残，則數(shù)據(jù)校驗(yàn)失敗

Authentication和Permission

Authentication指用戶認(rèn)證，Permission指權(quán)限機(jī)制为狸，這兩點(diǎn)是使RESTful API 強(qiáng)大歼郭、靈活和安全的基本保障。

URL Rules

Version your API

規(guī)范的API應(yīng)該包含版本信息辐棒，在RESTful API中病曾，最簡單的包含版本的方法是將版本信息放到url中，如：

/api/v1/posts/
/api/v1/drafts/

/api/v2/posts/
/api/v2/drafts/

另一種優(yōu)雅的做法是漾根，使用HTTP header中的accept來傳遞版本信息泰涂，這也是GitHub API 采取的策略。

Use nouns, not verbs

RESTful API 中的url是指向資源的辐怕，而不是描述行為的逼蒙，因此設(shè)計(jì)API時(shí)，應(yīng)使用名詞而非動(dòng)詞來描述語義寄疏，否則會(huì)引起混淆和語義不清是牢。

# Bad APIs
/api/getArticle/1/
/api/updateArticle/1/
/api/deleteArticle/1/

上面四個(gè)url都是指向同一個(gè)資源的僵井，雖然一個(gè)資源允許多個(gè)url指向它，但不同的url應(yīng)該表達(dá)不同的語義驳棱，上面的API可以優(yōu)化為：

# Good APIs
/api/Article/1/

article 資源的獲取批什、更新和刪除分別通過 GET, PUT 和 DELETE方法請(qǐng)求API即可。

Nested resources routing

如果要獲取一個(gè)資源子集蹈胡，采用 nested routing 是一個(gè)優(yōu)雅的方式渊季，如，列出所有文章中屬于Gevin編寫的文章

# List Gevin's articles
/api/authors/gevin/articles/

獲取資源子集的另一種方式是基于filter罚渐，這兩種方式都符合規(guī)范却汉，但語義不同：如果語義上將資源子集看作一個(gè)獨(dú)立的資源集合，則使用 nested routing 感覺更恰當(dāng)荷并，如果資源子集的獲取是出于過濾的目的合砂，則使用filter更恰當(dāng)。

Filter

對(duì)于資源集合源织，可以通過url參數(shù)對(duì)資源進(jìn)行過濾翩伪，如：

# List Gevin's articles
/api/articles?author=gevin

Pagination

對(duì)于資源集合，分頁獲取是一種比較合理的方式谈息。

Gevin的策略是缘屹，返回資源集合是，包含與分頁有關(guān)的數(shù)據(jù)如下：

{
  "page": 1,            # 當(dāng)前是第幾頁
  "pages": 3,           # 總共多少頁
  "per_page": 10,       # 每頁多少數(shù)據(jù)
  "has_next": true,     # 是否有下一頁數(shù)據(jù)
  "has_prev": false,    # 是否有前一頁數(shù)據(jù)
  "total": 27           # 總共多少數(shù)據(jù)
}

當(dāng)想API請(qǐng)求資源集合時(shí)侠仇，可選的分頁參數(shù)為：

另外逻炊，系統(tǒng)內(nèi)還設(shè)置一個(gè)per_page_max字段互亮，用于標(biāo)記系統(tǒng)允許的每頁最大記錄數(shù)，當(dāng)per_page值大于 per_page_max 值時(shí)余素，每頁記錄條數(shù)為 per_page_max豹休。

Url design tricks

• Url是區(qū)分大小寫的，這點(diǎn)經(jīng)常被忽略
• Back forward Slash桨吊，目前比較流行的API設(shè)計(jì)方案威根，通常建議url以/作為結(jié)尾
• 連接符-和下劃線，RESTful API 應(yīng)具備良好的可讀性视乐，當(dāng)url中某一個(gè)片段（segment）由多個(gè)單詞組成時(shí)洛搀，建議使用-來隔斷單詞，而不是使用

Flask RESTful API 開發(fā)基礎(chǔ)

Flask對(duì)HTTP方法的支持

Flask原生支持所有的HTTP方法

@app.route('/http-method-test/', methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE'])
def http_method_example():
    if request.method == 'GET':
        return 'Send request with `GET` method'
    elif request.method == 'POST':
        return 'Send request with `POST` method'
    elif request.method == 'PUT':
        return 'Send request with `PUT` method'
    elif request.method == 'PATCH':
        return 'Send request with `PATCH` method'
    elif request.method == 'DELETE':
        return 'Send request with `DELETE` method'

另外一種方式是采用Flask的MethodView

class HttpMethodExample(MethodView):
    def get(self):
        return 'Send request with `GET` method'

    def post(self):
        return 'Send request with `POST` method'

    def put(self):
        return 'Send request with `PUT` method'

    def patch(self):
        return 'Send request with `PATCH` method'

    def delete(self):
        return 'Send request with `DELETE` method'

app.add_url_rule('/http-method-test2/', view_func=HttpMethodExample.as_view('http_method_example2'))

Flask對(duì)序列化與反序列化的支持

序列化

RESTful API開發(fā)中的序列化炊林，通過包含了以下操作：

• 將Python native格式的數(shù)據(jù)（如dict和list）轉(zhuǎn)換為文本數(shù)據(jù)（如json或xml）
• 將文本數(shù)據(jù)作為請(qǐng)求的response返回給客戶端姥卢，response的http header里要同時(shí)附加application/json這個(gè)mimetype

這兩步操作，F(xiàn)lask提供的一個(gè)快捷函數(shù)jsonify()能直接完成

class SerializationExample(MethodView):
    def get(self):
        option = request.args.get('option')

        if option == 'list1':
            return self.test_list()
        if option == 'list2':
            return self.test_list2()
        if option == 'dict1':
            return self.test_dict1()
        if option == 'dict2':
            return self.test_dict2()
        if option == 'dict3':
            return self.test_dict3()


        msg = {
            'info': '`option` is needed in url as a url parameter',
            'avilable option values': 'list1, list2, test_dict1, test_dict2, test_dict2'
        }
        return jsonify(msg)



    def test_list(self):
        data = [{'a':1, 'b':2}, {'c':3, 'd':4}]
        return jsonify(result=data)

    def test_list2(self):
        data = [1,2,3,4,5,6,7,8]
        return jsonify(data)


    def test_dict1(self):
        data = {'a':1, 'b':2, 'c':3}
        return jsonify(data)

    def test_dict2(self):
        data = {'a':1, 'b':2, 'c':3}
        return jsonify(**data)

    def test_dict3(self):
        data = {'a':1, 'b':2, 'c':3}
        return jsonify(result=data)

app.add_url_rule('/serialization/', view_func=SerializationExample.as_view('serialization'))

反序列化

反序列化，即把文本形式的數(shù)據(jù)轉(zhuǎn)換為Python native類型數(shù)據(jù)的過程独榴。在RESTful API開發(fā)時(shí)僧叉，F(xiàn)lask內(nèi)置的get_json()方法，能夠把request中的json數(shù)據(jù)棺榔，轉(zhuǎn)換為Python標(biāo)準(zhǔn)庫中的dict或list瓶堕。

@app.route('/deserialization/', methods=['get', 'post'])
def deserialization():
    if request.method == 'POST':
        data = request.get_json()
        if not data:
            return 'No json data found', 400

        result = {
            'json data in request': data
        }
        return jsonify(result)

    return 'Please post json data'

Designing a RESTful API with Python and Flask

Implementing RESTful services in Python and Flask

Using the base Flask application we are now ready to implement the first entry point of our web service

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)

$ 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"
    }
  ]
}

Now let's write the second version of the GET method for our tasks resource.

from flask import abort

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

Here is how this function looks when invoked from 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>

We need to improve our 404 error handler.

from flask import make_response

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

And we get a much more API friendly error response

$ 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"
}

Next in our list is the POST method, which we will use to insert a new item in our task database.

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

To test this new function we can use the following curl command.

$ 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"
  }
}

The remaining two functions of our web service are shown below.

@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
    task = [task for task in tasks if task['id'] == task_id]
    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 = [task for task in tasks if task['id'] == task_id]
    if len(task) == 0:
        abort(404)
    tasks.remove(task[0])
    return jsonify({'result': True})

A function call that updates task #2 as being done would be done as follows.

$ 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"
    }
  ]
}

Securing a RESTful web service

The easiest way to secure our web service is to require clients to provide a username and a password.

There is a small Flask extension that can help with this, written by no other than yours truly. So let's go ahead and install Flask-HTTPAuth.

Let's say we want our web service to only be accessible to username miguel and password python. We can setup a Basic HTTP authentication as follows.

from flask_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)

With the authentication system setup, all that is left is to indicate which functions need to be protected, by adding the @auth.login_required decorator.

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

If we now try to invoke this function with curl this is what we get.

$ 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"
}

To be able to invoke this function we have to send our credentials.

$ 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"
    }
  ]
}

使用 Flask-RESTful 設(shè)計(jì) RESTful API

RESTful 服務(wù)器

[圖片上傳失敗...(image-fb5a62-1533821208427)]

這個(gè)服務(wù)唯一的資源叫做“任務(wù)”，它有如下一些屬性:

• id: 任務(wù)的唯一標(biāo)識(shí)符症歇。數(shù)字類型郎笆。
• title: 簡短的任務(wù)描述。字符串類型忘晤。
• description: 具體的任務(wù)描述宛蚓。文本類型钝尸。
• done: 任務(wù)完成的狀態(tài)谅辣。布爾值。

路由

Flask-RESTful 提供了一個(gè) Resource 基礎(chǔ)類带污，它能夠定義一個(gè)給定 URL 的一個(gè)或者多個(gè) HTTP 方法闰蛔。例如痕钢，定義一個(gè)可以使用 HTTP 的 GET, PUT 以及 DELETE 方法的 User 資源，你的代碼可以如下.

from flask import Flask
from flask.ext.restful import Api, Resource

app = Flask(__name__)
api = Api(app)

class UserAPI(Resource):
    def get(self, id):
        pass

    def put(self, id):
        pass

    def delete(self, id):
        pass

api.add_resource(UserAPI, '/users/<int:id>', endpoint = 'user')

add_resource 函數(shù)使用指定的 endpoint 注冊路由到框架上序六。如果沒有指定 endpoint任连，F(xiàn)lask-RESTful 會(huì)根據(jù)類名生成一個(gè)，但是有時(shí)候有些函數(shù)比如 url_for 需要 endpoint例诀，因此我會(huì)明確給 endpoint 賦值随抠。

我的待辦事項(xiàng) API 定義兩個(gè) URLs：/todo/api/v1.0/tasks（獲取所有任務(wù)列表），以及 /todo/api/v1.0/tasks/<int:id>（獲取單個(gè)任務(wù)）余佃。我們現(xiàn)在需要兩個(gè)資源

class TaskListAPI(Resource):
    def get(self):
        pass

    def post(self):
        pass

class TaskAPI(Resource):
    def get(self, id):
        pass

    def put(self, id):
        pass

    def delete(self, id):
        pass

api.add_resource(TaskListAPI, '/todo/api/v1.0/tasks', endpoint = 'tasks')
api.add_resource(TaskAPI, '/todo/api/v1.0/tasks/<int:id>', endpoint = 'task')

解析以及驗(yàn)證請(qǐng)求

當(dāng)我在以前的文章中實(shí)現(xiàn)此服務(wù)器的時(shí)候暮刃，我自己對(duì)請(qǐng)求的數(shù)據(jù)進(jìn)行驗(yàn)證跨算。

@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods = ['PUT'])
@auth.login_required
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': make_public_task(task[0]) } )

在這里, 我必須確保請(qǐng)求中給出的數(shù)據(jù)在使用之前是有效爆土，這樣使得函數(shù)變得又臭又長。

Flask-RESTful 提供了一個(gè)更好的方式來處理數(shù)據(jù)驗(yàn)證诸蚕，它叫做 RequestParser 類步势。這個(gè)類工作方式類似命令行解析工具 argparse。

首先背犯，對(duì)于每一個(gè)資源需要定義參數(shù)以及怎樣驗(yàn)證它們.

from flask.ext.restful import reqparse

class TaskListAPI(Resource):
    def __init__(self):
        self.reqparse = reqparse.RequestParser()
        self.reqparse.add_argument('title', type = str, required = True,
            help = 'No task title provided', location = 'json')
        self.reqparse.add_argument('description', type = str, default = "", location = 'json')
        super(TaskListAPI, self).__init__()

    # ...

class TaskAPI(Resource):
    def __init__(self):
        self.reqparse = reqparse.RequestParser()
        self.reqparse.add_argument('title', type = str, location = 'json')
        self.reqparse.add_argument('description', type = str, location = 'json')
        self.reqparse.add_argument('done', type = bool, location = 'json')
        super(TaskAPI, self).__init__()

    # ...

當(dāng)請(qǐng)求解析器被初始化坏瘩，解析和驗(yàn)證一個(gè)請(qǐng)求是很容易的。 例如漠魏，請(qǐng)注意 TaskAPI.put() 方法變的多么地簡單.

def put(self, id):
    task = filter(lambda t: t['id'] == id, tasks)
    if len(task) == 0:
        abort(404)
    task = task[0]
    args = self.reqparse.parse_args()
    for k, v in args.iteritems():
        if v != None:
            task[k] = v
    return jsonify( { 'task': make_public_task(task) } )

使用 Flask-RESTful 來處理驗(yàn)證的另一個(gè)好處就是沒有必要單獨(dú)地處理類似 HTTP 400 錯(cuò)誤倔矾，F(xiàn)lask-RESTful 會(huì)來處理這些。

生成響應(yīng)

原來設(shè)計(jì)的 REST 服務(wù)器使用 Flask 的 jsonify 函數(shù)來生成響應(yīng)。Flask-RESTful 會(huì)自動(dòng)地處理轉(zhuǎn)換成 JSON 數(shù)據(jù)格式.

return { 'task': make_public_task(task) }

Flask-RESTful 也支持自定義狀態(tài)碼哪自，如果有必要的話.

return { 'task': make_public_task(task) }, 201

Flask-RESTful 還有更多的功能丰包。make_public_task 能夠把來自原始服務(wù)器上的任務(wù)從內(nèi)部形式包裝成客戶端想要的外部形式。最典型的就是把任務(wù)的 id 轉(zhuǎn)成 uri壤巷。Flask-RESTful 就提供一個(gè)輔助函數(shù)能夠很優(yōu)雅地做到這樣的轉(zhuǎn)換邑彪，不僅僅能夠把 id 轉(zhuǎn)成 uri 并且能夠轉(zhuǎn)換其他的參數(shù).

from flask.ext.restful import fields, marshal

task_fields = {
    'title': fields.String,
    'description': fields.String,
    'done': fields.Boolean,
    'uri': fields.Url('task')
}

class TaskAPI(Resource):
    # ...

    def put(self, id):
        # ...
        return { 'task': marshal(task, task_fields) }

task_fields 結(jié)構(gòu)用于作為 marshal 函數(shù)的模板。fields.Uri 是一個(gè)用于生成一個(gè) URL 的特定的參數(shù)胧华。 它需要的參數(shù)是 endpoint寄症。

使用Flask設(shè)計(jì)RESTful的認(rèn)證

用戶數(shù)據(jù)庫

用戶的數(shù)據(jù)庫模型是十分簡單的。對(duì)于每一個(gè)用戶矩动，username 和 password_hash 將會(huì)被存儲(chǔ).

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key = True)
    username = db.Column(db.String(32), index = True)
    password_hash = db.Column(db.String(128))

出于安全原因有巧，用戶的原始密碼將不被存儲(chǔ)，密碼在注冊時(shí)被散列后存儲(chǔ)到數(shù)據(jù)庫中悲没。使用散列密碼的話剪决，如果用戶數(shù)據(jù)庫不小心落入惡意攻擊者的手里，他們也很難從散列中解析到真實(shí)的密碼檀训。

密碼決不能很明確地存儲(chǔ)在用戶數(shù)據(jù)庫中柑潦。

密碼散列

為了創(chuàng)建密碼散列，我將會(huì)使用 PassLib 庫峻凫，一個(gè)專門用于密碼散列的 Python 包渗鬼。

PassLib 提供了多種散列算法供選擇。custom_app_context 是一個(gè)易于使用的基于 sha256_crypt 的散列算法荧琼。

User 用戶模型需要增加兩個(gè)新方法來增加密碼散列和密碼驗(yàn)證功能.

from passlib.apps import custom_app_context as pwd_context

    class User(db.Model):
        # ...

        def hash_password(self, password):
            self.password_hash = pwd_context.encrypt(password)

        def verify_password(self, password):
            return pwd_context.verify(password, self.password_hash)

用戶注冊

個(gè)客戶端可以使用 POST 請(qǐng)求到 /api/users 上注冊一個(gè)新用戶譬胎。請(qǐng)求的主體必須是一個(gè)包含 username 和 password 的 JSON 格式的對(duì)象。

@app.route('/api/users', methods = ['POST'])
def new_user():
    username = request.json.get('username')
    password = request.json.get('password')
    if username is None or password is None:
        abort(400) # missing arguments
    if User.query.filter_by(username = username).first() is not None:
        abort(400) # existing user
    user = User(username = username)
    user.hash_password(password)
    db.session.add(user)
    db.session.commit()
    return jsonify({ 'username': user.username }), 201, {'Location': url_for('get_user', id = user.id, _external = True)}

這個(gè)函數(shù)是十分簡單地命锄。參數(shù) username 和 password 是從請(qǐng)求中攜帶的 JSON 數(shù)據(jù)中獲取堰乔，接著驗(yàn)證它們。

如果參數(shù)通過驗(yàn)證的話脐恩，新的 User 實(shí)例被創(chuàng)建镐侯。username 賦予給 User，接著使用 hash_password 方法散列密碼驶冒。用戶最終被寫入數(shù)據(jù)庫中苟翻。

這里是一個(gè)用戶注冊的請(qǐng)求，發(fā)送自 curl.

$ curl -i -X POST -H "Content-Type: application/json" -d '{"username":"miguel","password":"python"}' http://127.0.0.1:5000/api/users
HTTP/1.0 201 CREATED
Content-Type: application/json
Content-Length: 27
Location: http://127.0.0.1:5000/api/users/1
Server: Werkzeug/0.9.4 Python/2.7.3
Date: Thu, 28 Nov 2013 19:56:39 GMT

{
  "username": "miguel"
}

需要注意地是在真實(shí)的應(yīng)用中這里可能會(huì)使用安全的的 HTTP (譬如：HTTPS)骗污。如果用戶登錄的憑證是通過明文在網(wǎng)絡(luò)傳輸?shù)脑挸缑ǎ魏螌?duì) API 的保護(hù)措施是毫無意義的。

基于密碼的認(rèn)證

現(xiàn)在我們假設(shè)存在一個(gè)資源通過一個(gè) API 暴露給那些必須注冊的用戶需忿。這個(gè)資源是通過 URL: /api/resource 能夠訪問到诅炉。

為了保護(hù)這個(gè)資源蜡歹，我們將使用 HTTP 基本身份認(rèn)證，但是不是自己編寫完整的代碼來實(shí)現(xiàn)它涕烧，而是讓 Flask-HTTPAuth 擴(kuò)展來為我們做季稳。

使用 Flask-HTTPAuth，通過添加 login_required 裝飾器可以要求相應(yīng)的路由必須進(jìn)行認(rèn)證澈魄。

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

@app.route('/api/resource')
@auth.login_required
def get_resource():
    return jsonify({ 'data': 'Hello, %s!' % g.user.username })

能夠提供最大自由度的選擇(可能這也是唯一兼容 PassLib 散列)就是選用 verify_password 回調(diào)函數(shù)景鼠，這個(gè)回調(diào)函數(shù)將會(huì)根據(jù)提供的 username 和 password 的組合的，返回 True(通過驗(yàn)證) 或者 Flase(未通過驗(yàn)證)痹扇。Flask-HTTPAuth 將會(huì)在需要驗(yàn)證 username 和 password 對(duì)的時(shí)候調(diào)用這個(gè)回調(diào)函數(shù)铛漓。

verify_password 回調(diào)函數(shù)的實(shí)現(xiàn)如下

@auth.verify_password
def verify_password(username, password):
    user = User.query.filter_by(username = username).first()
    if not user or not user.verify_password(password):
        return False
    g.user = user
    return True

這個(gè)函數(shù)將會(huì)根據(jù) username 找到用戶，并且使用 verify_password() 方法驗(yàn)證密碼鲫构。如果認(rèn)證通過的話浓恶，用戶對(duì)象將會(huì)被存儲(chǔ)在 Flask 的 g 對(duì)象中，這樣視圖就能使用它结笨。

這里是用 curl 請(qǐng)求只允許注冊用戶獲取的保護(hù)資源包晰。

$ curl -u miguel:python -i -X GET http://127.0.0.1:5000/api/resource
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 30
Server: Werkzeug/0.9.4 Python/2.7.3
Date: Thu, 28 Nov 2013 20:02:25 GMT

{
  "data": "Hello, miguel!"
}

基于令牌的認(rèn)證

每次請(qǐng)求必須發(fā)送 username 和 password 是十分不方便，即使是通過安全的 HTTP 傳輸?shù)脑掃€是存在風(fēng)險(xiǎn)炕吸，因?yàn)榭蛻舳吮仨氁鎯?chǔ)不加密的認(rèn)證憑證伐憾，這樣才能在每次請(qǐng)求中發(fā)送。

一種基于之前解決方案的優(yōu)化就是使用令牌來驗(yàn)證請(qǐng)求赫模。

我們的想法是客戶端應(yīng)用程序使用認(rèn)證憑證交換了認(rèn)證令牌树肃，接下來的請(qǐng)求只發(fā)送認(rèn)證令牌。

令牌是具有有效時(shí)間瀑罗，過了有效時(shí)間后胸嘴，令牌變成無效，需要重新獲取新的令牌斩祭。令牌的潛在風(fēng)險(xiǎn)在于生成令牌的算法比較弱劣像，但是有效期較短可以減少風(fēng)險(xiǎn)。

有很多的方法可以加強(qiáng)令牌摧玫。一個(gè)簡單的強(qiáng)化方式就是根據(jù)存儲(chǔ)在數(shù)據(jù)庫中的用戶以及密碼生成一個(gè)隨機(jī)的特定長度的字符串耳奕，可能過期日期也在里面。令牌就變成了明文密碼的重排席赂，這樣就能很容易地進(jìn)行字符串對(duì)比吮铭，還能對(duì)過期日期進(jìn)行檢查时迫。

更加完善的實(shí)現(xiàn)就是不需要服務(wù)器端進(jìn)行任何存儲(chǔ)操作颅停，使用加密的簽名作為令牌。這種方式有很多的優(yōu)點(diǎn)掠拳，能夠根據(jù)用戶信息生成相關(guān)的簽名癞揉，并且很難被篡改。

Flask 使用類似的方式處理 cookies 的。這個(gè)實(shí)現(xiàn)依賴于一個(gè)叫做 itsdangerous 的庫喊熟，我們這里也會(huì)采用它柏肪。

令牌的生成以及驗(yàn)證將會(huì)被添加到 User 模型中，其具體實(shí)現(xiàn)如下芥牌。

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

class User(db.Model):
    # ...

    def generate_auth_token(self, expiration = 600):
        s = Serializer(app.config['SECRET_KEY'], expires_in = expiration)
        return s.dumps({ 'id': self.id })

    @staticmethod
    def verify_auth_token(token):
        s = Serializer(app.config['SECRET_KEY'])
        try:
            data = s.loads(token)
        except SignatureExpired:
            return None # valid token, but expired
        except BadSignature:
            return None # invalid token
        user = User.query.get(data['id'])
        return user

generate_auth_token() 方法生成一個(gè)以用戶 id 值為值烦味，’id’ 為關(guān)鍵字的字典的加密令牌。令牌中同時(shí)加入了一個(gè)過期時(shí)間壁拉，默認(rèn)為十分鐘(600 秒)谬俄。

驗(yàn)證令牌是在 verify_auth_token() 靜態(tài)方法中實(shí)現(xiàn)的。靜態(tài)方法被使用在這里弃理，是因?yàn)橐坏┝钆票唤獯a了用戶才可得知溃论。如果令牌被解碼了，相應(yīng)的用戶將會(huì)被查詢出來并且返回痘昌。

API 需要一個(gè)獲取令牌的新函數(shù)钥勋，這樣客戶端才能申請(qǐng)到令牌。

@app.route('/api/token')
@auth.login_required
def get_auth_token():
    token = g.user.generate_auth_token()
    return jsonify({ 'token': token.decode('ascii') })

注意：這個(gè)函數(shù)是使用了 auth.login_required 裝飾器辆苔，也就是說需要提供 username 和 password算灸。

剩下來的就是決策客戶端怎樣在請(qǐng)求中包含這個(gè)令牌。

HTTP 基本認(rèn)證方式不特別要求 usernames 和 passwords 用于認(rèn)證驻啤，在 HTTP 頭中這兩個(gè)字段可以用于任何類型的認(rèn)證信息乎婿。基于令牌的認(rèn)證街佑，令牌可以作為 username 字段谢翎，password 字段可以忽略。

這就意味著服務(wù)器需要同時(shí)處理 username 和 password 作為認(rèn)證沐旨，以及令牌作為 username 的認(rèn)證方式森逮。verify_password 回調(diào)函數(shù)需要同時(shí)支持這兩種方式。

@auth.verify_password
def verify_password(username_or_token, password):
    # first try to authenticate by token
    user = User.verify_auth_token(username_or_token)
    if not user:
        # try to authenticate with username/password
        user = User.query.filter_by(username = username_or_token).first()
        if not user or not user.verify_password(password):
            return False
    g.user = user
    return True

新版的 verify_password 回調(diào)函數(shù)會(huì)嘗試認(rèn)證兩次磁携。首先它會(huì)把 username 參數(shù)作為令牌進(jìn)行認(rèn)證褒侧。如果沒有驗(yàn)證通過的話，就會(huì)像基于密碼認(rèn)證的一樣谊迄，驗(yàn)證 username 和 password闷供。

如下的 curl 請(qǐng)求能夠獲取一個(gè)認(rèn)證的令牌。

$ curl -u miguel:python -i -X GET http://127.0.0.1:5000/api/token
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 139
Server: Werkzeug/0.9.4 Python/2.7.3
Date: Thu, 28 Nov 2013 20:04:15 GMT

{
  "token": "eyJhbGciOiJIUzI1NiIsImV4cCI6MTM4NTY2OTY1NSwiaWF0IjoxMzg1NjY5MDU1fQ.eyJpZCI6MX0.XbOEFJkhjHJ5uRINh2JA1BPzXjSohKYDRT472wGOvjc"
}

現(xiàn)在可以使用令牌獲取資源统诺。

$ curl -u eyJhbGciOiJIUzI1NiIsImV4cCI6MTM4NTY2OTY1NSwiaWF0IjoxMzg1NjY5MDU1fQ.eyJpZCI6MX0.XbOEFJkhjHJ5uRINh2JA1BPzXjSohKYDRT472wGOvjc:unused -i -X GET http://127.0.0.1:5000/api/resource
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 30
Server: Werkzeug/0.9.4 Python/2.7.3
Date: Thu, 28 Nov 2013 20:05:08 GMT

{
  "data": "Hello, miguel!"
}

How to structure a Flask-RESTPlus web service for production builds

Flask-RESTPlus

Flask-RESTPlus is an extension for Flask that adds support for quickly building REST APIs. Flask-RESTPlus encourages best practices with minimal setup. It provides a coherent collection of decorators and tools to describe your API and expose its documentation properly (using Swagger).

Project Setup and Organization

In the project directory, create a new package called app. Inside app, create two packages main and test. Your directory structure should look similar to the one below.

Inside the main package, create three more packages namely: controller, service and model. The model package will contain all of our database models while the service package will contain all the business logic of our application and finally the controller package will contain all our application endpoints.

[圖片上傳失敗...(image-1dc2bf-1533821208427)]

Configuration Settings

In the main package create a file called config.py with the following content.

import os

# uncomment the line below for postgres database url from environment variable
# postgres_local_base = os.environ['DATABASE_URL']

basedir = os.path.abspath(os.path.dirname(__file__))


class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'my_precious_secret_key')
    DEBUG = False


class DevelopmentConfig(Config):
    # uncomment the line below to use postgres
    # SQLALCHEMY_DATABASE_URI = postgres_local_base
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class TestingConfig(Config):
    DEBUG = True
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
    PRESERVE_CONTEXT_ON_EXCEPTION = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class ProductionConfig(Config):
    DEBUG = False
    # uncomment the line below to use postgres
    # SQLALCHEMY_DATABASE_URI = postgres_local_base


config_by_name = dict(
    dev=DevelopmentConfig,
    test=TestingConfig,
    prod=ProductionConfig
)

key = Config.SECRET_KEY

The configuration file contains three environment setup classes which includes testing, development, and production.

We will be using the application factory pattern for creating our Flask object. This pattern is most useful for creating multiple instances of our application with different settings. This facilitates the ease at which we switch between our testing, development and production environment by calling the create_app function with the required parameter.

In the __init__.py file inside the main package, enter the following lines of code.

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt

from .config import config_by_name

db = SQLAlchemy()
flask_bcrypt = Bcrypt()


def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config_by_name[config_name])
    db.init_app(app)
    flask_bcrypt.init_app(app)

    return app

Flask Script

Now let’s create our application entry point. In the root directory of the project, create a file called manage.py with the following content.

import os
import unittest

from flask_migrate import Migrate, MigrateCommand
from flask_script import Manager

from app import blueprint
from app.main import create_app, db
from app.main.model import user, blacklist

app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')
app.register_blueprint(blueprint)

app.app_context().push()

manager = Manager(app)

migrate = Migrate(app, db)

manager.add_command('db', MigrateCommand)


@manager.command
def run():
    app.run()


@manager.command
def test():
    """Runs the unit tests."""
    tests = unittest.TestLoader().discover('app/test', pattern='test*.py')
    result = unittest.TextTestRunner(verbosity=2).run(tests)
    if result.wasSuccessful():
        return 0
    return 1

if __name__ == '__main__':
    manager.run()

The above code within manage.py does the following

• line 4 and 5 imports the migrate and manager modules respectively (we will be using the migrate command soon).

• line 9 calls the create_app function we created initially to create the application instance with the required parameter from the environment variable which can be either of the following - dev, prod, test. If none is set in the environment variable, the default dev is used.

• line 13 and 15 instantiates the manager and migrate classes by passing the app instance to their respective constructors.

• In line 17,we pass the db and MigrateCommandinstances to the add_command interface of the managerto expose all the database migration commands through Flask-Script.

• line 20 and 25 marks the two functions as executable from the command line.

Flask-Migrate exposes two classes, Migrate and MigrateCommand. The Migrateclass contains all the functionality of the extension. The MigrateCommand class is only used when it is desired to expose database migration commands through the Flask-Script extension.

Database Models and Migration

Now let’s create our models. We will be using the db instance of the sqlalchemy to create our models.

The db instance contains all the functions and helpers from both sqlalchemy and sqlalchemy.orm and it provides a class called Model that is a declarative base which can be used to declare models.

In the model package, create a file called user.py with the following content.

from .. import db, flask_bcrypt
import datetime
from app.main.model.blacklist import BlacklistToken
from ..config import key
import jwt


class User(db.Model):
    """ User Model for storing user related details """
    __tablename__ = "user"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    email = db.Column(db.String(255), unique=True, nullable=False)
    registered_on = db.Column(db.DateTime, nullable=False)
    admin = db.Column(db.Boolean, nullable=False, default=False)
    public_id = db.Column(db.String(100), unique=True)
    username = db.Column(db.String(50), unique=True)
    password_hash = db.Column(db.String(100))

    @property
    def password(self):
        raise AttributeError('password: write-only field')

    @password.setter
    def password(self, password):
        self.password_hash = flask_bcrypt.generate_password_hash(password).decode('utf-8')

    def check_password(self, password):
        return flask_bcrypt.check_password_hash(self.password_hash, password)


    def encode_auth_token(self, user_id):
        """
        Generates the Auth Token
        :return: string
        """
        try:
            payload = {
                'exp': datetime.datetime.utcnow() + datetime.timedelta(days=1, seconds=5),
                'iat': datetime.datetime.utcnow(),
                'sub': user_id
            }
            return jwt.encode(
                payload,
                key,
                algorithm='HS256'
            )
        except Exception as e:
            return e

    @staticmethod
    def decode_auth_token(auth_token):
        """
        Decodes the auth token
        :param auth_token:
        :return: integer|string
        """
        try:
            payload = jwt.decode(auth_token, key)
            is_blacklisted_token = BlacklistToken.check_blacklist(auth_token)
            if is_blacklisted_token:
                return 'Token blacklisted. Please log in again.'
            else:
                return payload['sub']
        except jwt.ExpiredSignatureError:
            return 'Signature expired. Please log in again.'
        except jwt.InvalidTokenError:
            return 'Invalid token. Please log in again.'

    def __repr__(self):
        return "<User '{}'>".format(self.username)

The above code within user.py does the following

• line 3: The user class inherits from db.Model class which declares the class as a model for sqlalchemy.

• line 7 through 13 creates the required columns for the user table.

• line 21 is a setter for the field password_hash and it uses flask-bcrypt to generate a hash using the provided password.

• line 24 compares a given password with already savedpassword_hash.

Now to generate the database table from the user model we just created, we will use migrateCommand through the manager interface. For managerto detect our models, we will have to import theuser model by adding below code to manage.py file.

from app.main.model import user

Testing

Configuration

Create a file called test_config.py in the test package with the content below.

import os
import unittest

from flask import current_app
from flask_testing import TestCase

from manage import app
from app.main.config import basedir


class TestDevelopmentConfig(TestCase):
    def create_app(self):
        app.config.from_object('app.main.config.DevelopmentConfig')
        return app

    def test_app_is_development(self):
        self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
        self.assertTrue(app.config['DEBUG'] is True)
        self.assertFalse(current_app is None)
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
        )


class TestTestingConfig(TestCase):
    def create_app(self):
        app.config.from_object('app.main.config.TestingConfig')
        return app

    def test_app_is_testing(self):
        self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
        self.assertTrue(app.config['DEBUG'])
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
        )


class TestProductionConfig(TestCase):
    def create_app(self):
        app.config.from_object('app.main.config.ProductionConfig')
        return app

    def test_app_is_production(self):
        self.assertTrue(app.config['DEBUG'] is False)


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

Run the test using the command below

python manage.py test

User Operations

Now let’s work on the following user related operations:

• creating a new user
• getting a registered user with his public_id
• getting all registered users.

User Service class: This class handles all the logic relating to the user model.

In the service package, create a new file user_service.py with the following content

import uuid
import datetime

from app.main import db
from app.main.model.user import User


def save_new_user(data):
    user = User.query.filter_by(email=data['email']).first()
    if not user:
        new_user = User(
            public_id=str(uuid.uuid4()),
            email=data['email'],
            username=data['username'],
            password=data['password'],
            registered_on=datetime.datetime.utcnow()
        )
        save_changes(new_user)
        return generate_token(new_user)
    else:
        response_object = {
            'status': 'fail',
            'message': 'User already exists. Please Log in.',
        }
        return response_object, 409


def get_all_users():
    return User.query.all()


def get_a_user(public_id):
    return User.query.filter_by(public_id=public_id).first()


def generate_token(user):
    try:
        # generate the auth token
        auth_token = user.encode_auth_token(user.id)
        response_object = {
            'status': 'success',
            'message': 'Successfully registered.',
            'Authorization': auth_token.decode()
        }
        return response_object, 201
    except Exception as e:
        response_object = {
            'status': 'fail',
            'message': 'Some error occurred. Please try again.'
        }
        return response_object, 401


def save_changes(data):
    db.session.add(data)
    db.session.commit()

The above code within user_service.py does the following

• line 8 through 29 creates a new user by first checking if the user already exists; it returns a success response_object if the user doesn’t exist else it returns an error code 409 and a failure response_object.

• line 33 and 37 return a list of all registered users and a user object by providing the public_id respectively.

• line 40 to 42 commits the changes to database.

No need to use jsonify for formatting an object to JSON, Flask-restplus does it automatically

In the main package, create a new package called util . This package will contain all the necessary utilities we might need in our application.

In the util package, create a new file dto.py. As the name implies, the data transfer object (DTO) will be responsible for carrying data between processes. In our own case, it will be used for marshaling data for our API calls. We will understand this better as we proceed.

from flask_restplus import Namespace, fields


class UserDto:
    api = Namespace('user', description='user related operations')
    user = api.model('user', {
        'email': fields.String(required=True, description='user email address'),
        'username': fields.String(required=True, description='user username'),
        'password': fields.String(required=True, description='user password'),
        'public_id': fields.String(description='user Identifier')
    })


class AuthDto:
    api = Namespace('auth', description='authentication related operations')
    user_auth = api.model('auth_details', {
        'email': fields.String(required=True, description='The email address'),
        'password': fields.String(required=True, description='The user password '),
    })

The above code within dto.py does the following

• line 5 creates a new namespace for user related operations. Flask-RESTPlus provides a way to use almost the same pattern as Blueprint. The main idea is to split your app into reusable namespaces. A namespace module will contain models and resources declaration.

• line 6 creates a new user dto through the model interface provided by the api namespace in line 5.

User Controller: The user controller class handles all the incoming HTTP requests relating to the user.

Under the controller package, create a new file called user_controller.py with the following content.

from flask import request
from flask_restplus import Resource

from app.main.util.decorator import admin_token_required
from ..util.dto import UserDto
from ..service.user_service import save_new_user, get_all_users, get_a_user

api = UserDto.api
_user = UserDto.user


@api.route('/')
class UserList(Resource):
    @api.doc('list_of_registered_users')
    @admin_token_required
    @api.marshal_list_with(_user, envelope='data')
    def get(self):
        """List all registered users"""
        return get_all_users()

    @api.expect(_user, validate=True)
    @api.response(201, 'User successfully created.')
    @api.doc('create a new user')
    def post(self):
        """Creates a new User """
        data = request.json
        return save_new_user(data=data)


@api.route('/<public_id>')
@api.param('public_id', 'The User identifier')
@api.response(404, 'User not found.')
class User(Resource):
    @api.doc('get a user')
    @api.marshal_with(_user)
    def get(self, public_id):
        """get a user given its identifier"""
        user = get_a_user(public_id)
        if not user:
            api.abort(404)
        else:
            return user

line 1 through 8 imports all the required resources for the user controller.

We defined two concrete classes in our user controller which are userList and user. These two classes extends the abstract flask-restplus resource.

Concrete resources should extend from this class and expose methods for each supported HTTP method. If a resource is invoked with an unsupported HTTP method, the API will return a response with status 405 Method Not Allowed. Otherwise the appropriate method is called and passed all arguments from the URL rule used when adding the resource to an API instance.

The api namespace in line 7 above provides the controller with several decorators which includes but is not limited to the following:

• api.route: A decorator to route resources

• api.marshal_with: A decorator specifying the fields to use for serialization (This is where we use the userDto we created earlier)

• api.marshal_list_with: A shortcut decorator for marshal_with above with as_list = True

• api.doc: A decorator to add some api documentation to the decorated object

• api.response: A decorator to specify one of the expected responses

• api.expect: A decorator to Specify the expected input model ( we still use the userDto for the expected input)

• api.param: A decorator to specify one of the expected parameters

We have now defined our namespace with the user controller. Now its time to add it to the application entry point.

In the __init__.py file of app package, enter the following.

from flask_restplus import Api
from flask import Blueprint

from .main.controller.user_controller import api as user_ns
from .main.controller.auth_controller import api as auth_ns

blueprint = Blueprint('api', __name__)

api = Api(blueprint,
          title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
          version='1.0',
          description='a boilerplate for flask restplus web service'
          )

api.add_namespace(user_ns, path='/user')
api.add_namespace(auth_ns)

The above code within blueprint.py does the following.

• In line 8, we create a blueprint instance by passing name and import_name. API is the main entry point for the application resources and hence needs to be initialized with the blueprint in line 10.

• In line 16 , we add the user namespace user_ns to the list of namespaces in the API instance.

We have now defined our blueprint. It’s time to register it on our Flask app.

Update manage.py by importing blueprint and registering it with the Flask application instance.

Now open the URL http://127.0.0.1:5000 in your browser. You should see the swagger documentation.

Security and Authentication

Let’s create a model blacklistToken for storing blacklisted tokens. In the models package, create a blacklist.py file with the following content.

from .. import db
import datetime


class BlacklistToken(db.Model):
    """
    Token Model for storing JWT tokens
    """
    __tablename__ = 'blacklist_tokens'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    token = db.Column(db.String(500), unique=True, nullable=False)
    blacklisted_on = db.Column(db.DateTime, nullable=False)

    def __init__(self, token):
        self.token = token
        self.blacklisted_on = datetime.datetime.now()

    def __repr__(self):
        return '<id: token: {}'.format(self.token)

    @staticmethod
    def check_blacklist(auth_token):
        # check whether auth token has been blacklisted
        res = BlacklistToken.query.filter_by(token=str(auth_token)).first()
        if res:
            return True
        else:
            return False

Lets not forget to migrate the changes to take effect on our database.

Import the blacklist class in manage.py.

from app.main.model import blacklist

Next create blacklist_service.py in the service package with the following content for blacklisting a token.

from app.main import db

from app.main.model.blacklist import BlacklistToken


def save_token(token):
    blacklist_token = BlacklistToken(token=token)
    try:
        # insert the token
        db.session.add(blacklist_token)
        db.session.commit()
        response_object = {
            'status': 'success',
            'message': 'Successfully logged out.'
        }
        return response_object, 200
    except Exception as e:
        response_object = {
            'status': 'fail',
            'message': e
        }
        return response_object, 200

Update the user model with two static methods for encoding and decoding tokens. Add the following imports.

import datetime
import jwt
from app.main.model.blacklist import BlacklistToken
from ..config import key

Encoding

def encode_auth_token(self, user_id):
        """
        Generates the Auth Token
        :return: string
        """
        try:
            payload = {
                'exp': datetime.datetime.utcnow() + datetime.timedelta(days=1, seconds=5),
                'iat': datetime.datetime.utcnow(),
                'sub': user_id
            }
            return jwt.encode(
                payload,
                key,
                algorithm='HS256'
            )
        except Exception as e:
            return e

Decoding: Blacklisted token, expired token and invalid token are taken into consideration while decoding the authentication token.

  @staticmethod
  def decode_auth_token(auth_token):
        """
        Decodes the auth token
        :param auth_token:
        :return: integer|string
        """
        try:
            payload = jwt.decode(auth_token, key)
            is_blacklisted_token = BlacklistToken.check_blacklist(auth_token)
            if is_blacklisted_token:
                return 'Token blacklisted. Please log in again.'
            else:
                return payload['sub']
        except jwt.ExpiredSignatureError:
            return 'Signature expired. Please log in again.'
        except jwt.InvalidTokenError:
            return 'Invalid token. Please log in again.'

Now let’s write a test for the user model to ensure that our encode and decode functions are working properly.

In the test package, create base.py file with the following content

from flask_testing import TestCase

from app.main import db
from manage import app


class BaseTestCase(TestCase):
    """ Base Tests """

    def create_app(self):
        app.config.from_object('app.main.config.TestingConfig')
        return app

    def setUp(self):
        db.create_all()
        db.session.commit()

    def tearDown(self):
        db.session.remove()
        db.drop_all()

The BaseTestCase sets up our test environment ready before and after every test case that extends it.

Create test_user_medol.py with the following test cases.

import unittest

import datetime

from app.main import db
from app.main.model.user import User
from app.test.base import BaseTestCase


class TestUserModel(BaseTestCase):

    def test_encode_auth_token(self):
        user = User(
            email='test@test.com',
            password='test',
            registered_on=datetime.datetime.utcnow()
        )
        db.session.add(user)
        db.session.commit()
        auth_token = user.encode_auth_token(user.id)
        self.assertTrue(isinstance(auth_token, bytes))

    def test_decode_auth_token(self):
        user = User(
            email='test@test.com',
            password='test',
            registered_on=datetime.datetime.utcnow()
        )
        db.session.add(user)
        db.session.commit()
        auth_token = user.encode_auth_token(user.id)
        self.assertTrue(isinstance(auth_token, bytes))
        self.assertTrue(User.decode_auth_token(auth_token.decode("utf-8") ) == 1)


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

Run the test with python manage.py test. All the tests should pass.

Let’s create the authentication endpoints for login and logout.

• First we need a dto for the login payload. We will use the auth dto for the @expect annotation in login endpoint. Add the code below to the dto.py

class AuthDto:
    api = Namespace('auth', description='authentication related operations')
    user_auth = api.model('auth_details', {
        'email': fields.String(required=True, description='The email address'),
        'password': fields.String(required=True, description='The user password '),
    })

• Next, we create an authentication helper class for handling all authentication related operations. This auth_helper.py will be in the service package and will contain two static methods which are login_user and logout_user

When a user is logged out, the user’s token is blacklisted ie the user can’t log in again with that same token.

from app.main.model.user import User
from ..service.blacklist_service import save_token


class Auth:

    @staticmethod
    def login_user(data):
        try:
            # fetch the user data
            user = User.query.filter_by(email=data.get('email')).first()
            if user and user.check_password(data.get('password')):
                auth_token = user.encode_auth_token(user.id)
                if auth_token:
                    response_object = {
                        'status': 'success',
                        'message': 'Successfully logged in.',
                        'Authorization': auth_token.decode()
                    }
                    return response_object, 200
            else:
                response_object = {
                    'status': 'fail',
                    'message': 'email or password does not match.'
                }
                return response_object, 401

        except Exception as e:
            print(e)
            response_object = {
                'status': 'fail',
                'message': 'Try again'
            }
            return response_object, 500

    @staticmethod
    def logout_user(data):
        if data:
            auth_token = data.split(" ")[1]
        else:
            auth_token = ''
        if auth_token:
            resp = User.decode_auth_token(auth_token)
            if not isinstance(resp, str):
                # mark the token as blacklisted
                return save_token(token=auth_token)
            else:
                response_object = {
                    'status': 'fail',
                    'message': resp
                }
                return response_object, 401
        else:
            response_object = {
                'status': 'fail',
                'message': 'Provide a valid auth token.'
            }
            return response_object, 403

Let us now create endpoints for login and logout operations.

In the controller package, create auth_controller.py with the following contents.

from flask import request
from flask_restplus import Resource

from app.main.service.auth_helper import Auth
from ..util.dto import AuthDto

api = AuthDto.api
user_auth = AuthDto.user_auth


@api.route('/login')
class UserLogin(Resource):
    """
        User Login Resource
    """
    @api.doc('user login')
    @api.expect(user_auth, validate=True)
    def post(self):
        # get the post data
        post_data = request.json
        return Auth.login_user(data=post_data)


@api.route('/logout')
class LogoutAPI(Resource):
    """
    Logout Resource
    """
    @api.doc('logout a user')
    def post(self):
        # get auth token
        auth_header = request.headers.get('Authorization')
        return Auth.logout_user(data=auth_header)

At this point the only thing left is to register the auth api namespace with the application Blueprint
Update __init__.py file of app package with the following

# app/__init__.py
from flask_restplus import Api
from flask import Blueprint

from .main.controller.user_controller import api as user_ns
from .main.controller.auth_controller import api as auth_ns

blueprint = Blueprint('api', __name__)

api = Api(blueprint,
          title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
          version='1.0',
          description='a boilerplate for flask restplus web service'
          )

api.add_namespace(user_ns, path='/user')
api.add_namespace(auth_ns)

Run the application with python manage.py run and open the url http://127.0.0.1:5000 in your browser.

Before we write some tests to ensure our authentication is working as expected, let’s modify our registration endpoint to automatically login a user once the registration is successful.

Add the method generate_token below to user_service.py.

def generate_token(user):
    try:
        # generate the auth token
        auth_token = user.encode_auth_token(user.id)
        response_object = {
            'status': 'success',
            'message': 'Successfully registered.',
            'Authorization': auth_token.decode()
        }
        return response_object, 201
    except Exception as e:
        response_object = {
            'status': 'fail',
            'message': 'Some error occurred. Please try again.'
        }
        return response_object, 401

The generate_token method generates an authentication token by encoding the user id. This token is the returned as a response.

Next, replace the return block in save_new_user method below

response_object = {
    'status': 'success',
    'message': 'Successfully registered.'
}
return response_object, 201

with

return generate_token(new_user)

Now its time to test the login and logout functionalities. Create a new test file test_auth.py in the test package with the following content.

import unittest

from app.main import db
from app.main.model.blacklist import BlacklistToken
import json
from app.test.base import BaseTestCase


def register_user(self):
    return self.client.post(
        '/user/',
        data=json.dumps(dict(
            email='joe@gmail.com',
            username='username',
            password='123456'
        )),
        content_type='application/json'
    )


def login_user(self):
    return self.client.post(
        '/auth/login',
        data=json.dumps(dict(
            email='joe@gmail.com',
            password='123456'
        )),
        content_type='application/json'
    )


class TestAuthBlueprint(BaseTestCase):
    def test_registration(self):
        """ Test for user registration """
        with self.client:
            response = register_user(self)
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'success')
            self.assertTrue(data['message'] == 'Successfully registered.')
            self.assertTrue(data['Authorization'])
            self.assertTrue(response.content_type == 'application/json')
            self.assertEqual(response.status_code, 201)

    def test_registered_with_already_registered_user(self):
        """ Test registration with already registered email"""
        register_user(self)
        with self.client:
            response = register_user(self)
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'fail')
            self.assertTrue(
                data['message'] == 'User already exists. Please Log in.')
            self.assertTrue(response.content_type == 'application/json')
            self.assertEqual(response.status_code, 409)

    def test_registered_user_login(self):
        """ Test for login of registered-user login """
        with self.client:
            # user registration
            resp_register = register_user(self)
            data_register = json.loads(resp_register.data.decode())
            self.assertTrue(data_register['status'] == 'success')
            self.assertTrue(
                data_register['message'] == 'Successfully registered.'
            )
            self.assertTrue(data_register['Authorization'])
            self.assertTrue(resp_register.content_type == 'application/json')
            self.assertEqual(resp_register.status_code, 201)
            # registered user login
            response = login_user(self)
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'success')
            self.assertTrue(data['message'] == 'Successfully logged in.')
            self.assertTrue(data['Authorization'])
            self.assertTrue(response.content_type == 'application/json')
            self.assertEqual(response.status_code, 200)

    def test_non_registered_user_login(self):
        """ Test for login of non-registered user """
        with self.client:
            response = login_user(self)
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'fail')
            print(data['message'])
            self.assertTrue(data['message'] == 'email or password does not match.')
            self.assertTrue(response.content_type == 'application/json')
            self.assertEqual(response.status_code, 401)

    def test_valid_logout(self):
        """ Test for logout before token expires """
        with self.client:
            # user registration
            resp_register = register_user(self)
            data_register = json.loads(resp_register.data.decode())
            self.assertTrue(data_register['status'] == 'success')
            self.assertTrue(
                data_register['message'] == 'Successfully registered.')
            self.assertTrue(data_register['Authorization'])
            self.assertTrue(resp_register.content_type == 'application/json')
            self.assertEqual(resp_register.status_code, 201)
            # user login
            resp_login = login_user(self)
            data_login = json.loads(resp_login.data.decode())
            self.assertTrue(data_login['status'] == 'success')
            self.assertTrue(data_login['message'] == 'Successfully logged in.')
            self.assertTrue(data_login['Authorization'])
            self.assertTrue(resp_login.content_type == 'application/json')
            self.assertEqual(resp_login.status_code, 200)
            # valid token logout
            response = self.client.post(
                '/auth/logout',
                headers=dict(
                    Authorization='Bearer ' + json.loads(
                        resp_login.data.decode()
                    )['Authorization']
                )
            )
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'success')
            self.assertTrue(data['message'] == 'Successfully logged out.')
            self.assertEqual(response.status_code, 200)

    def test_valid_blacklisted_token_logout(self):
        """ Test for logout after a valid token gets blacklisted """
        with self.client:
            # user registration
            resp_register = register_user(self)
            data_register = json.loads(resp_register.data.decode())
            self.assertTrue(data_register['status'] == 'success')
            self.assertTrue(
                data_register['message'] == 'Successfully registered.')
            self.assertTrue(data_register['Authorization'])
            self.assertTrue(resp_register.content_type == 'application/json')
            self.assertEqual(resp_register.status_code, 201)
            # user login
            resp_login = login_user(self)
            data_login = json.loads(resp_login.data.decode())
            self.assertTrue(data_login['status'] == 'success')
            self.assertTrue(data_login['message'] == 'Successfully logged in.')
            self.assertTrue(data_login['Authorization'])
            self.assertTrue(resp_login.content_type == 'application/json')
            self.assertEqual(resp_login.status_code, 200)
            # blacklist a valid token
            blacklist_token = BlacklistToken(
                token=json.loads(resp_login.data.decode())['Authorization'])
            db.session.add(blacklist_token)
            db.session.commit()
            # blacklisted valid token logout
            response = self.client.post(
                '/auth/logout',
                headers=dict(
                    Authorization='Bearer ' + json.loads(
                        resp_login.data.decode()
                    )['Authorization']
                )
            )
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'fail')
            self.assertTrue(data['message'] == 'Token blacklisted. Please log in again.')
            self.assertEqual(response.status_code, 401)


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

Route protection and Authorization

So far, we have successfully created our endpoints, implemented login and logout functionalities but our endpoints remains unprotected.

We need a way to define rules that determines which of our endpoint is open or requires authentication or even an admin privilege.

We can achieve this by creating custom decorators for our endpoints.

Before we can protect or authorize any of our endpoints, we need to know the currently logged in user. We can do this by pulling the Authorization token from the header of the current request by using the flask library request.We then decode the user details from the Authorization token.

In the Auth class of auth_helper.py file, add the following static method

@staticmethod
def get_logged_in_user(new_request):
        # get the auth token
        auth_token = new_request.headers.get('Authorization')
        if auth_token:
            resp = User.decode_auth_token(auth_token)
            if not isinstance(resp, str):
                user = User.query.filter_by(id=resp).first()
                response_object = {
                    'status': 'success',
                    'data': {
                        'user_id': user.id,
                        'email': user.email,
                        'admin': user.admin,
                        'registered_on': str(user.registered_on)
                    }
                }
                return response_object, 200
            response_object = {
                'status': 'fail',
                'message': resp
            }
            return response_object, 401
        else:
            response_object = {
                'status': 'fail',
                'message': 'Provide a valid auth token.'
            }
            return response_object, 401

Now that we can retrieve the logged in user from the request, let’s go ahead and create the decorators.

Create a file decorator.py in the util package with the following content

from functools import wraps

from flask import request

from app.main.service.auth_helper import Auth


def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):

        data, status = Auth.get_logged_in_user(request)
        token = data.get('data')

        if not token:
            return data, status

        return f(*args, **kwargs)

    return decorated


def admin_token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):

        data, status = Auth.get_logged_in_user(request)
        token = data.get('data')

        if not token:
            return data, status

        admin = token.get('admin')
        if not admin:
            response_object = {
                'status': 'fail',
                'message': 'admin token required'
            }
            return response_object, 401

        return f(*args, **kwargs)

    return decorated

For more information about decorators and how to create them, take a look at this link.

Now that we have created the decorators token_required and admin_token_required for valid token and for an admin token respectively, all that is left is to annotate the endpoints which we wish to protect with the freecodecamp orgappropriate decorator.

電影訂票系統(tǒng)后臺(tái)開發(fā)REST API 和 Swagger UI

github歪脏，點(diǎn)擊此處

創(chuàng)建資源

# *-* coding: utf-8 *-*
from flask import request
from app.utils import UUID
from app.models import Favorite, Movie, db
from flask_restplus import Namespace, Resource
from flask_login import current_user, login_required

api = Namespace('favorite', description='收藏模塊')


@api.route('/')
class FavoritesResource(Resource):
    @login_required
    def get(self):
        """獲取收藏列表(需登錄)"""
        return [f.__json__() for f in current_user.favorites], 200

    @api.doc(parser=api.parser().add_argument(
      'movieId', type=str, required=True, help='電影id', location='form')
    )
    @login_required
    def post(self):
        """收藏電影(需登錄)"""
        mid = request.form.get('movieId', '')
        movie = Movie.query.get(mid)
        if movie is None:
            return {'message': '電影不存在'}, 233
        movie = current_user.favorites.filter_by(movieId=mid).first()
        if movie is not None:
            return {'message': '不能重復(fù)收藏同部電影'}, 233

        favorite = Favorite()
        favorite.id = UUID()
        favorite.username = current_user.id
        favorite.movieId = mid
        db.session.add(favorite)
        db.session.commit()

        return {'message': '收藏成功', 'id': favorite.id}, 200


@api.route('/<id>')
@api.doc(params={'id': '收藏id'})
class FavoriteResource(Resource):
    @login_required
    def delete(self, id):
        """取消收藏(需登錄)"""
        favorite = current_user.favorites.filter_by(id=id).first()
        if favorite is None:
            return {'message': '您沒有這個(gè)收藏'}, 233
        db.session.delete(favorite)
        db.session.commit()
        return {'message': '取消收藏成功'}, 200

創(chuàng)建了資源之后，只需要進(jìn)行初始化即可實(shí)現(xiàn) RESTful 服務(wù)粮呢。

# *-* coding: utf-8 *-*
from flask import Flask
from flask_restplus import Api
from favorite import api as ns1

api = Api(
    title='MonkeyEye',
    version='1.0',
    description='猿眼電影訂票系統(tǒng)API',
    doc='/swagger/',             # Swagger UI: http://localhost:5000/swagger/
    catch_all_404s=True,
    serve_challenge_on_401=True
)

api.add_namespace(ns1, path='/api/favorites')

app = Flask(__name__)
api.init_app(app)

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

參考

1. Eve. The Simple Way to REST婿失，點(diǎn)擊此處
2. Building-Serverless-Python-Web-Services-with-Zappa钞艇，點(diǎn)擊此處
3. 使用 Flask 設(shè)計(jì) RESTful APIs，英文版豪硅，點(diǎn)擊此處哩照，中文版，點(diǎn)擊此處
4. flask-mongorest懒浮，點(diǎn)擊此處
5. Buidling a database driven RESTFUL JSON API in Python 3 with Flask Flask-Restful and SQLAlchemy飘弧，點(diǎn)擊此處
6. Building beautiful REST APIs using Flask, Swagger UI and Flask-RESTPlus，點(diǎn)擊此處
7. Creating Flask RESTful API Using Python & MySQL砚著，點(diǎn)擊此處
8. How to structure a Flask-RESTPlus web service for production builds眯牧，點(diǎn)擊此處