以 Flask / MongoDB 製作 通訊錄 CRUD

程式碼太多，請參考本文的專案 https://github.com/shinder/flask-practice，以下只列出部份程式碼：

主程式：

import modules.address_book

app.add_url_rule('/address-book/list/', None, modules.address_book.ab_list)
app.add_url_rule('/address-book/list/<int:page>', None, modules.address_book.ab_list)
app.add_url_rule('/address-book/edit/<_id>', None, modules.address_book.ab_edit_get, methods=['GET'])
app.add_url_rule('/address-book/edit', None, modules.address_book.ab_edit_post, methods=['POST'])
app.add_url_rule('/address-book/add', None, modules.address_book.ab_add_get, methods=['GET'])
app.add_url_rule('/address-book/add', None, modules.address_book.ab_add_post, methods=['POST'])
app.add_url_rule('/address-book/delete/<_id>', None, modules.address_book.ab_delete)

app/modules/address_book.py 的內容：

from flask import Flask, request, render_template, session, jsonify, redirect
from bson.objectid import ObjectId
import json
import math
import re
import modules.mongo_connection

email_pattern = r"^([\w-]+(?:\.[\w-]+)*)@((?:[\w-]+\.)*\w[\w-]{0,66})\.([a-z]{2,6}(?:\.[a-z]{2})?)$"

def ab_list(page=1):
    (db, connection) = modules.mongo_connection.getDB('test')
    output = {
        'page_name': 'ab-list',
        'page_title': '列表 - 通訊錄',
        'totalRows': 0,   # 總筆數
        'perPage': 3,     # 每一頁最多幾筆
        'totalPages': 0,  # 總頁數
        'page': page,     # 用戶要查看的頁數
        'rows': [],       # 當頁的資料
    }
    output['totalRows']  = db.address_book.count_documents({})
    output['totalPages'] = math.ceil(output['totalRows']/output['perPage'])
    output['page'] = 1 if page < 1 else page
    if output['page'] > output['totalPages']:
        output['page'] = output['totalPages']
    if output['totalRows']==0:
        output['rows'] = []
    else:
        cursor = db.address_book.find({},
                        sort=[('_id', -1)],
                        skip=(output['page']-1) * output['perPage'],
                        limit=output['perPage']
                    )
        for doc in cursor:
            doc['_id'] = str(doc['_id'])
            output['rows'].append(doc)
    return render_template('address-book/list.html', **output)

def ab_edit_get(_id):
    (db, connection) = modules.mongo_connection.getDB('test')
    try:
        oid = ObjectId(_id)
    except:
        return redirect("/address-book/list/1", code=302)

    row = db.address_book.find_one({'_id': oid})
    row['_id'] = _id  # 使用字串
    if not row:
        return redirect("/address-book/list/1", code=302)
    else:
        row['page_name'] = 'ab_edit'
        row['page_title'] = '修改 - 通訊錄'
        return render_template('address-book/edit.html', **row)

def ab_edit_post():
    output = {
        'success': False,
        'error': '',
    }
    if len(request.form.get('name')) < 2:
        output['error'] = '姓名字元長度太短'
        return output

    email_match = re.search(email_pattern, request.form.get('email'), re.I)
    if not email_match:
        output['error'] = 'Email 格式錯誤'
        return output

    (db, connection) = modules.mongo_connection.getDB('test')
    doc = request.form.to_dict()
    _id = doc['_id']
    del doc['_id']
    rr = db.address_book.replace_one({'_id': ObjectId(_id)}, doc)
    # print(rr) # pymongo.results.UpdateResult
    if rr.modified_count==1:
        output['success'] = True
    else:
        output['error'] = '資料沒有變更';

    return output

def ab_add_get():
    output = {
        'page_name': 'ab_add',
        'page_title': '新增 - 通訊錄',
    }
    return render_template('address-book/add.html', **output)

def ab_add_post():
    output = {
        'success': False,
        'error': '',
    }
    if len(request.form.get('name')) < 2:
        output['error'] = '姓名字元長度太短'
        return output

    email_match = re.search(email_pattern, request.form.get('email'), re.I)
    if not email_match:
        output['error'] = 'Email 格式錯誤'
        return output

    (db, connection) = modules.mongo_connection.getDB('test')
    doc = request.form.to_dict()
    rr = db.address_book.insert_one(doc)
    # print(dir(rr))  # InsertOneResult
    if rr.inserted_id:
        output['success'] = True
    else:
        output['error'] = '資料沒有新增';
    return output

def ab_delete(_id):
    (db, connection) = modules.mongo_connection.getDB('test')
    try:
        oid = ObjectId(_id)
    except:
        return redirect("/address-book/list/1", code=302)

    rr = db.address_book.delete_one({'_id': oid})
    # print(dir(rr))  # DeleteResult
    referer = request.headers.get('referer')
    if not referer:
        return redirect("/address-book/list/1", code=302)
    else:
        return redirect(referer, code=302)

Flask 連線 MongoDB

本文的參考專案 https://github.com/shinder/flask-practice

這裡直接用最基本的 PyMongo 連線方式，官方文件。

建立連線模組，app/modules/mongo_connection.py：

from pymongo import MongoClient

url = 'mongodb://localhost:27017'
client = None

def getDB(db_name):
    global client
    if not client:
        client = MongoClient(url)
    return (client[db_name], client)

這個 MongoDB 連線模組的想法和 mysql_connection.py 的想法相同，有使用才會連線。測試的 route 部份：

import modules.mongo_connection

@app.route('/try-mongo')
def try_mongo():
    (db, c) = modules.mongo_connection.getDB('test')
    one = db.inventory.find_one()
    one['_id'] = str(one['_id'])  # 將 ObjectId 轉換為字串顯示
    return one

Flask 新增資料到 MySQL

本文的參考專案 https://github.com/shinder/flask-practice

承上篇，這裡要使用 Postman POST 傳送 JSON 文件，然後 Flask 接收後寫入資料庫。

MySQL 官網新增資料的範例

Flask route 寫法：

@app.route('/receive-json', methods=['POST'])
def receive_json():
    (cursor, cnx) = modules.mysql_connection.get_cursor()
    data = json.loads(request.get_data())  # JSON 字串轉換為 dict
    p = {}
    sids = []  # 用來記錄新增的 primary key
    p['name'] = data['name'] if 'name' in data else ''
    p['email'] = data['email'] if 'email' in data else ''
    p['mobile'] = data['mobile'] if 'mobile' in data else ''
    p['birthday'] = data['birthday'] if 'birthday' in data else '1900-01-01'
    p['address'] = data['address'] if 'address' in data else ''

    # 兩種作法
    sql1 = ("INSERT INTO `address_book`"
        "(`name`, `email`, `mobile`, `birthday`, `address`, `created_at`"
        ") VALUES (%s, %s, %s, %s, %s, NOW())")
    sql2 = ("INSERT INTO `address_book`"
        "(`name`, `email`, `mobile`, `birthday`, `address`, `created_at`"
        ") VALUES (%(name)s, %(email)s, %(mobile)s, %(birthday)s, %(address)s, NOW())")

    cursor.execute(sql1, (p['name'], p['email'], p['mobile'], p['birthday'], p['address']))
    sids.append(cursor.lastrowid)  # 取得新增項目的 primary key
    cursor.execute(sql2, p)  # 使用 dict
    sids.append(cursor.lastrowid)
    cnx.commit()  # 提交新增的資料才會生效

    return jsonify(sids)  # 輸出 JSON 格式

Postman 發需求的網址 http://localhost:5000/receive-json，JSON 文件如下：

{
    "address": "台南市",
    "birthday": "2000-11-22",
    "email": "wwww@test.com",
    "mobile": "0918777-777",
    "name": "陳小華"
}

Flask 使用 MySQL Connector

本文的參考專案 https://github.com/shinder/flask-practice

專案資料表 address_book 參考

連線 MySQL DB 的套件，這邊介紹最陽春的，就是 MySQL 官方出的 mysql-connector。可以依照 mysql-connector 開發人員指引 介紹的方式安裝。用 pip 安裝應該是最簡單的：

pip install mysql-connector

連線的功能我們把它獨立出來成為一個模組 app/modules/mysql_connection.py，其中 get_cursor() 可以同時回傳游標物件和連線物件：

import mysql.connector

connect_data = {
    'host': 'localhost',
    'user': 'root',
    'passwd': 'root',
    'database': 'test'
}
cnx = None

def get_connection():
    global cnx  # 將連線物件存放在全域變數
    if not cnx:
        cnx = mysql.connector.connect(**connect_data)
        return cnx
    else:
        return cnx

def get_cursor():
    cursor = get_connection().cursor(dictionary=True)  # 讀出資料使用 dict，預設為 tuple
    return (cursor, get_connection())  # 同時回傳 cursor 和 connection

在主檔案定義 route：

import modules.mysql_connection

@app.route('/try-mysql')
def try_mysql():
    (cursor, cnx) = modules.mysql_connection.get_cursor()
    sql = ("SELECT * FROM address_book")
    cursor.execute(sql)
    return render_template('data_table.html', t_data=cursor.fetchall())

樣版檔 app/templates/data_table.html：

    <tbody>
        {% for i in t_data %}
        <tr>
            <td>{{ i.name }}</td>
            <td>{{ i.email }}</td>
        </tr>
        {% endfor %}
    </tbody>

Flask 內建的 session 功能

本文的參考專案 https://github.com/shinder/flask-practice

• 使用 session 時，需要設定 cookie 的加密字串（secret key）。
• 取得 session 值，使用 session.get('鍵')。
• 設定 session 值，使用 session['鍵']=值。

from flask import session
from datetime import timedelta
import os

app.config['SECRET_KEY'] = os.urandom(24)
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(minutes=60)

@app.route('/try-session')
def try_session():
    if not session.get('what'):
        session['what'] = 1
    else:
        session['what'] += 1
    return str( session.get('what') )

Flask 路由路徑參數

本文的參考專案 https://github.com/shinder/flask-practice

• 路徑參數，可以使用角括號標示，如下範例的 action 和 id。
• 參數也可以指定類型，如 int。
• 可將多個路由定義疊起來，使用相同的函式處理，並能使用預設值。

@app.route('/params/')
@app.route('/params/<action>/')
@app.route('/params/<action>/<int:id>')
def my_params(action='none', id=0):
    return ( {
        'action': action,
        'id': id
    } )

Flask 接收表單資料及 query string

本文的參考專案 https://github.com/shinder/flask-practice

• request.args 可取得 GET 參數（query string）
• request.form 可取得 POST 參數（表單資料）
• request.files 可取得表單上傳的檔案

以下為取得 GET 和 POST 參數的例子：

@app.route('/try-qs')
def queryString():
    # query string 轉成 dict
    # http://localhost:5000/try-qs?a[]=1&b=34&a[]=5
    output = {
        'args': request.args,
        'a[]': request.args.getlist('a[]'),
        'get_b': request.args.get('b'),
        'get_a[]': request.args.get('a[]'),
    }
    return output

@app.route('/try-post', methods=['POST'])  # 限定使用 POST
def try_post():
    # 表單資料 urlencoded, form-data 皆可, 使用 postman 測試
    output = {
        'form': request.form,
        'a[]': request.form.getlist('a[]'),
        'post_b': request.form.get('b'),
        'post_a[]': request.form.get('a[]'),
    }
    return output

@app.route('/try-post2', methods=['POST'])
def try_post2():
    # 使用 postman post json 資料: {"a":11,"b":22}
    output = {
        'content_type': request.content_type,
        'data': request.data.decode('utf-8'),
        'json': request.get_json(),
    }
    return output

測試時，/try-post 和 /try-post2 可以使用 postman 測試。 getlist() 能取得所有相同名稱的參數，而拿到 list 類型的物件（陣列）。但不會將名稱有帶中括號的參數自動轉換為陣列。

使用 postman 將 json 文件 post 給 /try-post2 路由。request.data.decode('utf-8') 可以拿到字串；request.get_json() 可以拿到由 JSON 字串轉換而成的 dict 或 list。

Flask 的樣版系統 Jinja

本文的參考專案 https://github.com/shinder/flask-practice

Flask 樣版說明可以參考 這裡

在主程式加入以下片段：

from flask import render_template


@app.route('/basic-template')
def basic_template():
    return render_template('basic.html', name='是在哈囉', age=25)

@app.route('/basic-template2')
def basic_template2():
    output = {
        'name': '小明',
        'age': 30
    }
    return render_template('basic.html', ** output)

render_template() 用來呈現頁面，第一個參數為樣版檔（樣版檔必須放在 app/templates 資料夾內），第二個參數之後為欲傳入樣版的資料。

如 basic_template2() 也可以將資料包成 dict 然後再傳入。

樣版 app/templates/basic.html 的內容如下：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>{{ name }}</title>
</head>
<body>
<h2>Hello, {{ name }}</h2>
<h6>{{ age }}</h6>
</body>
</html>

Flask 靜態檔案資料夾

在專案的資料夾的 app 內建立 static 資料夾，並在建立 app 物件時給第二個參數，指定路徑。

app = Flask(__name__, '/')

static 資料夾就會看成類似 apache 的 document root。

將處理的方法放到不同的 .py 檔

本文的參考專案 https://github.com/shinder/flask-practice

將處理的方法寫到 modules/functions.py 裡

from flask import request  # 滙入 request


def show_cookies():
    return request.cookies

在主要檔案 main.py 滙入：

import modules.functions

# 語法：add_url_rule(rule, endpoint=None, view_func=None, provide_automatic_options=None, **options)
app.add_url_rule('/show-cookies', 'show-cookies', modules.functions.show_cookies)
# app.add_url_rule('/show-cookies', 'show-cookies')
# app.view_functions['show-cookies'] = modules.functions.show_cookies

• 使用 add_url_rule() 可以將處理函式設定給路由。
• endpoint 用來做處理函式及路由的對應。給 None 時，則使用函式名稱。
• 這樣的方式可以將處理函式放在不同的檔案，以方便維護。

將 request headers 寫入檔案

本文的參考專案 https://github.com/shinder/flask-practice

from flask import request  # 滙入 request
import json

@app.route('/save-headers')
def save_headers():
    dict1 = {
        'cookies': {}
    }
    for i in request.headers:
        print(i)  # 查看取出的 headers 資料
        dict1[i[0]] = i[1]
    # 查看 cookies
    for i in request.cookies:
        dict1['cookies'][i] = request.cookies[i]
    file1 = open('headers.json', 'w')
    file1.write(json.dumps(dict1))  # 存成 JSON
    return dict1

• dict1 為用來暫存資料的 dict，並設定好結構
• request.headers 用 for/in 取出，為 tuple
• request.cookies 為 dict
• json 為預設套件，dumps 為轉換為 JSON 字串
• return dict 會自動轉換為 JSON 格式
• 拜訪 localhost:5000/save-headers 可以看到結果

建立 Flask 專案

建立 Flask 專案

本文的參考專案 https://github.com/shinder/flask-practice

首先如之前的文章 建立專案。

md flask-practice #建立專案資料夾
cd flask-practice #到專案目錄
python3 -m venv venv #安裝虛擬環境
source venv/bin/activate #啟動虛擬環境(mac)

安裝 Flask：

pip install flask

可以使用下列兩個命列中的一個查看安裝的套件：

pip list
pip freeze

建立 app 資料夾，用來存放自己撰寫的程式，並在裡面建立 main.py

from flask import Flask

# __name__ 用來 application 的相對位置
# 若是直接啟動的程式 __name__ 為 '__main__'
# 若是被滙入， __name__ 會是被滙入的名稱
app = Flask(__name__)

# decorators 後面定義的 function 會變成 decorators 的參數
# 類似 JavaScript 的 callback function
@app.route('/')
def index():
    return '<h2>哈囉 Flask</h2>'

在專案目錄建立執行的 run.sh，在 mac 記得將檔案屬性設定為 +x 可執行：

# mac
export FLASK_APP=app/main.py
export FLASK_ENV=development
flask run --host=localhost --port=5000

# windows
# set FLASK_APP=app/hello.py
# set FLASK_ENV=development
# flask run

在 terminal 執行 run.sh，並在瀏覽器拜訪 localhost:5000 即可看到我們的第一個頁面。