以 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)
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)
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})
referer = request.headers.get('referer')
if not referer:
return redirect("/address-book/list/1", code=302)
else:
return redirect(referer, code=302)