軟體|應用|使用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
 
執行成功
 
開啟瀏覽器,輸入以下網址
http://192.168.100.210:5000
 
 
[建立 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;開啟終端機
curl -i http://192.168.100.210:5000/todo/api/v1.0/tasks

以下為回應內容
--------------------
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
 
執行
 
curl -i http://192.168.100.210:5000/todo/api/v1.0/tasks/2
 
 
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
執行
curl -i http://192.168.100.210:5000/todo/api/v1.0/tasks/2

尚未 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"
  }
}


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

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/<int:task_id>', 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/<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])})


@app.route('/todo/api/v1.0/tasks/<int:task_id>', 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 -i http://192.168.100.210:5000/todo/api/v1.0/tasks

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

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

發表迴響

你的電子郵件位址並不會被公開。 必要欄位標記為 *