IT｜軟體｜應用｜使用Flask實現一個RESTful API Service in 樹莓派

REST 六特性
  • Client-Server：服務器端與客戶端分離。
  • Stateless（無狀態）：每次客戶端請求必需包含完整的信息，換句話說，每一次請求都是獨立的。
  • Cacheable（可緩存）：服務器端必需指定哪些請求是可以緩存的。
  • Layered System（分層結構）：服務器端與客戶端通訊必需標準化，服務器的變更並不會影響客戶端。
  • Uniform Interface（統一接口）：客戶端與服務器端的通訊方法必需是統一的。
  • Code on demand（按需執行代碼）：服務器端可以在上下文中執行代碼或者腳本。

RESTful API Service 的樣子
REST架構就是為了HTTP協議設計的。RESTful API Service 的核心概念是管理資源。資源是由 URIs 來表示，客戶端使用 HTTP 當中的 POST、OPTIONS、GET、PUT、DELETE 等方法發送請求到服務器，改變相應的資源狀態。


HTTP請求方法通常也十分合適去描述操作資源的動作：
HTTP 方法
動作例子
GET獲取資源信息
http://example.com/api/orders（ 檢索訂單清單）
GET
獲取資源信息
http://example.com/api/orders/123 （ 檢索訂單 #123）
POST
創建一個資源
http://example.com/api/orders（ 使用帶數據的請求，創建一個新的訂單）
PUT
更新一個資源
http://example.com/api/orders/123 （ 使用帶數據的請求，更新#123訂單）
DELETE
刪除一個資源
http://example.com/api/orders/123（刪除訂單#123）

[設計一個點單的 API Service]

對於我的監控項目，四種請求都要用到，首先設定URL

第一步
http://hostname/monitor/api/v1.0
上面的 URL包括了應用程序的名稱、API 版本，既提供了命名空間的劃分，同時又與其它系統區分開來。版本號在升級新特性時十分有用，當一個新功能特性增加在新版本下面時，不至於影響舊版本。

第二步，規劃資源的URL
                                                  資源：https://eliyar.biz/restful-api-server-by-flask-python/
照片信息包括：
                   * id：唯一標示，整形。
                   * picname：照片文件名字，字符串。
                   * time：照片拍攝時間，字符串。
                   * status：狀態（1：正常，2：可疑），整形。
用戶信息包括：
                   * id：唯一標示，整形。
                   * username：用戶名，字符串。
                   * password：密碼，字符串（加密算法暫不考慮）。
                   * email：郵箱，字符串。

Flask 介紹
Flask是一個使用 Python 編寫的輕量級 Web 應用框架。其 WSGI 工具箱採用 Werkzeug ，模板引擎則使用 Jinja2 。

[安裝 Flask in 樹莓派]

安裝
更新系統
sudo apt-get update
sudo apt-get -y dist-upgrade

安裝 Python3 若以安裝，此步驟可以省略
sudo apt-get -y install python3-pip

安裝 Flask
------------------
sudo pip3 install flask

安裝 Flask 安全性模組（稍後會用到）
------------------
sudo pip3 install flask-httpauth

設定 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(host='127.0.0.1', debug=False) —> 可設置為 True

——————————
如果你希望暫時先讓所有人都可以連的話, 把127.0.0.1改成0.0.0.0 即可.


修改可執行權限；啟動 Web Server
chmod a+x app.py
python3 ./app.py

執行成功

開啟瀏覽器，輸入以下網址


[建立 Web Service]


現在有一個 get_tasks 的函數, 能透過 URL: http:192.168.100.210:5000/todo/api/v1.0/tasks 來拜訪(資料傳輸的方式為JSON)，但是限定只能透過Get的方式來存取。

Get 不帶參數
task.py
#!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(host='0.0.0.0', debug=False)

修改權限；啟動 Web Server
chmod a+x task.py
python3 ./task.py


透過 curl 來測試 Web service；開啟終端機

以下為回應內容
--------------------
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 317
Server: Werkzeug/0.11.15 Python/3.5.3
Date: Mon, 16 Apr 2018 07:46:59 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"
    }
  ]
}


Get 帶參數
接著是另外一種形式. 透過 Get 所帶的參數來決定回傳什麼樣的內容

task2.py
#!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})

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]})



if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False)

修改權限；啟動 Web Server
chmod a+x task2.py
python3 ./task2.py

執行



404 制定統一錯誤訊息
那有時候會不小心發生404的狀況. 當然http本身就會吐自身的error回去. 但如果你希望符合一致性也透過JSON回傳的話. 可以使用Flask errorhandler來處理.

task3.py
#!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})

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]})

from flask import make_response

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

if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False)

修改可執行權限；啟動 Web Server
chmod a+x task3.py
python3 ./task3.py

執行

curl -i http://192.168.100.210:5000/todo/api/v1.0/tasks/3


POST
接下來就是 POST 方法，我們用來在我們的任務數據庫中插入一個新的任務

task4.py

#!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})

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]})

from flask import make_response

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


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

if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False)

修改可執行權限；啟動 Web Server
chmod a+x task4.py
python3 ./task4.py

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


PUT/DELETE

Put & Delete 的情況, 作者則合併一起講, 可能是比較少用到的吧. (我自己平常也只用Get&Post而已, 看來之後得改一下, 萬惡的Post啊...)

task5.py
#!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})


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]})

from flask import make_response

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


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

@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})

@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]})

if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False)

修改可執行權限；啟動 Web Server
chmod a+x task5.py
python3 ./task5.py

執行

尚未 PUT 前，第二筆的數據資料
-----------------
Kevins-MacBook-Pro:~ Kevin$ curl -i http://192.168.100.210:5000/todo/api/v1.0/tasks/2
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 152
Server: Werkzeug/0.11.15 Python/3.5.3
Date: Tue, 17 Apr 2018 03:59:43 GMT

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

執行 PUT
-----------------
curl -i -H "Content-Type: application/json" -X PUT -d '{"done":true}' http://192.168.100.210:5000/todo/api/v1.0/tasks/2

PUT 結果
-----------------
Kevins-MacBook-Pro:~ Kevin$ curl -i -H "Content-Type: application/json" -X PUT -d '{"done":true}' http://192.168.100.210:5000/todo/api/v1.0/tasks/2
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 151
Server: Werkzeug/0.11.15 Python/3.5.3
Date: Tue, 17 Apr 2018 04:00:37 GMT

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



PUT 後，第二筆的數據資料
-----------------
Kevins-MacBook-Pro:~ Kevin$ curl -i http://192.168.100.210:5000/todo/api/v1.0/tasks/2
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 151
Server: Werkzeug/0.11.15 Python/3.5.3
Date: Tue, 17 Apr 2018 04:01:55 GMT

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



[RESTful API Service 安全性]

 Flask module: Flask-HTTPAuth 強化 RESTful API Service 安全性

task6.py
 #!flask/bin/python
from flask import Flask, jsonify, abort, request, make_response, url_for
from flask.ext.httpauth import HTTPBasicAuth

app = Flask(__name__, static_url_path="")
auth = HTTPBasicAuth()


@auth.get_password
def get_password(username):
    if username == 'pi':
        return '999999'
    return None


@auth.error_handler
def unauthorized():
    return make_response(jsonify({'error': 'Unauthorized access'}), 403)
    # return 403 instead of 401 to prevent browsers from displaying the default auth dialog


@app.errorhandler(400)
def not_found(error):
    return make_response(jsonify({'error': 'Bad request'}), 400)


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


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


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


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


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


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


@app.route('/todo/api/v1.0/tasks/&lt;int:task_id&gt;', 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])})


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


if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=False)


修改可執行權限；啟動 Web Server
chmod a+x task6.py
python3 ./task6.py

執行
未帶入驗證資訊
------------

代入驗證資訊
------------
curl -u pi:999999 -i http://192.168.100.210:5000/todo/api/v1.0/tasks

[參考]
[1].使用Flask實現一個RESTful API Server

留言

張貼留言

這個網誌中的熱門文章

IoT｜硬體｜樹莓派｜外接麥克風及喇叭設置

圖片
輸出喇叭與麥克風設定 [軟體安裝] 安裝  ALSA (Advanced Linux Sound Architecture)  的音效驅動程式 sudo apt-get install alsa-base alsa-utils [接線] 喇叭接樹莓派自身3.5mm音源孔 音效卡接 USB，外接麥克風的3.5mm接頭插到音效卡上  以樹莓派內建3.5mm音源孔作為聲音輸出 sudo raspi-config → Advanced options → Audio → 選擇Force 3.5mm ('headphone') jack 檢查是否有抓到音效卡 lsusb 樹莓派桌面進行如下設定，並分別將 BCM2835_ALSA 及 USB PnP Sounde device 分別設為 default PCM 及default Microphone。 上述設定完成後請將樹莓派重新開機 測試喇叭是否正確播放聲音測試喇吧是否有聲音,測試指令，它會依左聲道、右聲道的順序反覆播放，可以按「Ctrl+C」來停止 speaker-test -c 2 -t wav -D hw:0,0 [細部設定] 若沒有聲音，確認喇叭是否接在內建音效卡為0的位置上，必且將聲音調整到最大 alsamixer #card 0 是內建的 BCM2835 晶片 #card 1 就是外接的 USB 音效卡 調整麥克風，Speaker 表示喇叭音量、Mic 表示麥克風直接輸出到喇叭的音量、Auto Gain Control 表示自動調整麥克風增益 F6->選則 sound card 按 esc 離開 將剛剛編號0號及1號音效卡設定儲存下來 sudo alsactl store 0 sudo alsactl store 1 列出樹莓派內建預設的音效卡（編號0）內容 amixer --card 0 contents 列出樹莓派內建預設的音效卡（編號1）內容 amixer -
閱讀完整內容

成長｜語文｜學習-英文 持續更新！

圖片
主格/受格/所有格    2018-01-18 週四 八大詞性   2018-03-04 v. →動詞  n. →名詞  adj. →形容詞  adv. →副詞  con. →連接詞  prep. →介系詞  pron. →代名詞  aux. →助詞  英文縮寫    2017-10-17 週二     臺南市立大灣高中影音教學頻道    2018-03-23 週五  [八大詞類] 動詞 使役動詞 Part 1 使役動詞 Part 2 使役動詞 Part 3 使役動詞 Part 4 使役動詞 Part 5 關係詞 高中英文文法-關係詞01 - 概論 代名詞： 含有 it 的句型 1 含有 it 的句型 2 含有 it 的句型 3 含有 it 的句型 4 含有 it 的句型 5 不定詞： 高中英文文法 - 不定詞 01 分詞： 英文文法-分詞-01 簡介 英文文法-分詞-02 分詞的位置      英文文法-分詞-03分詞的主動與被動      主詞與動詞一致 主詞語動詞的一致 Part I 主詞語動詞的一致 Part II 主詞與動詞的一致 Part III 主詞與動詞的一致 Part IV 主詞與動詞的一致 Part V [五大基本句型] 英文文法 - 五大基本句型 英文文法 - 五大基本句型 - Exercise 01 五大基本句型 - exercise 02 [子句] 形容詞子句 Part 1 六個符號多益函授    2017-08-23週三 快速準備：/Users/Kevin/iBook/英語學習/六個符號/六個符號-多益函授/影片/快速準備 文法     https://youtu.be/SDkqRF1AAm8 片語（介詞＋名詞） 子句 [ ] S+V of/at/in/due to/because of/as a result of 形容詞子句：which（物）/who（人）= that although（雖然）/when（當）/if（是否）/why（
閱讀完整內容

IoT｜硬體｜通訊｜Arduino 使用 SoftwareSerial Library 與電腦通訊

圖片
介紹 UART Arduino  硬件在 0 pin  和 1 pin  口支持串行通信（ 電腦通過  USB  連接） ，其實是使用了一個 ( 內置芯片 ) 稱為  UART  的芯片。這種硬件允許  ATmega  即使正在進行其他任務時也能通信，只要 serial 的 64 byte 緩存有其他的空間。 SoftwareSerial 使用  SoftwareSerial Library  可以在  Arduino  的任何數字引腳上進行通信，因此軟件端口數可以擴展，並且波特率可達 115200bps 。 有個叫 Mikal Hart 的傢伙編寫了 NewSoftSerial library ，這個庫改進了 SoftwareSerial Library ，優點更多，因此之後的版本都是基於這個庫的。這種其實是軟件模擬通信的，具有很多局限性。 範例程式：選擇不同 baud rate 與設備通訊 #include <SoftwareSerial.h> SoftwareSerial DLSerial(10, 11); // RX,TX int cammand; int baudChoose; void setup() {   Serial.begin(9600);   DLSerial.begin(9600);    } void loop() {   String s = "";   while (Serial.available()) {     char c = Serial.read();     if (c != '\n') {       s += c;     }     delay(5);    // 沒有延遲的話 UART 串口速度會跟不上Arduino的速度，會導致資料不完整   }   String value;   if (s != "") {     cammand = (s.substring(0, 2)).toInt();     value  = (s.substring(3, 22));     switch (c
閱讀完整內容