在前一篇完成重構的 serverless 平台 v2 版中, 我將所有應用會用到的金鑰與全權杖等集中於系統目錄下的環境變數檔 .env 統一管理; 本篇則是要在此基礎上為此 serverless 函式執行平台添加呼叫統計功能, 為此也修改了主程式動態載入函式模組時傳遞給 main() 函式的參數結構, 添加了一個 protected 參數來傳送管理模組名稱串列給函式模組.
本系列全部測試文章參考 :
呼叫紀錄儲存在平台根目錄 ~/flask_apps/serverless 下的一個名為 serverless.db 的 SQLite 資料庫檔案, 其用法參考 :
本篇只會在 severless.db 中建立並維護一個兩欄位 (func_name 與 call_count) 的資料表 call_stats, 用來記錄被呼叫的函式模組名稱與累加的被呼叫次數. 取名為 serverless.db 是為了保留未來功能擴充性.
1. 修改主程式 serverless.py :
主要是增加 sqlite3 模組之匯入, 第一次執行時的資料庫初始化函式, 以及每次函式模組被呼叫前的統計增量函式 :
# serverless.py
from flask import Flask, request, jsonify, session
import importlib.util
import os
import logging
from dotenv import dotenv_values
import sqlite3
# 指定呼叫統計資料庫位置 (必須在 init_db() 定義之前)
DB_PATH='./serverless.db'
def check_auth(): # 檢查使用者是否已登入
return session.get('authenticated') == True
def init_db(): # 初始化呼叫紀錄資料庫
if not os.path.exists(DB_PATH):
conn=sqlite3.connect(DB_PATH)
cursor=conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS call_stats (
func_name TEXT PRIMARY KEY,
call_count INTEGER NOT NULL
)
""")
conn.commit()
conn.close()
def record_call(func_name): # 紀錄函式呼叫次數
if not os.path.exists(DB_PATH): # 若資料庫檔不存在就建立
init_db()
try:
conn=sqlite3.connect(DB_PATH)
cursor=conn.cursor()
cursor.execute('SELECT call_count FROM call_stats WHERE func_name=?', (func_name,))
row=cursor.fetchone()
if row: # 有找到 : 呼叫次數增量 1
cursor.execute('UPDATE call_stats SET call_count=call_count + 1 WHERE func_name=?', (func_name,))
else: # 沒找到 : 第一次呼叫設為 1
cursor.execute('INSERT INTO call_stats (func_name, call_count) VALUES (?, 1)', (func_name,))
conn.commit()
conn.close()
except Exception as e:
logging.error(f'Failed to record call stats for {func_name}: {e}')
app=Flask(__name__)
# 初始化資料庫
init_db()
# 從 .env 讀取權杖 (密碼) 與金鑰
config=dotenv_values('.env')
SECRET_TOKEN=config.get('SECRET_TOKEN') # 易記的令牌 (類似密碼)
SECRET_KEY=config.get('SECRET_KEY') # 簽章加密用的金鑰
app.secret_key=SECRET_KEY # 用來簽章與驗證 session cookie
# 指定函式模組所在的資料夾
FUNCTIONS_DIR=os.path.expanduser('./functions')
# 指定錯誤日誌檔 (在目前工作目錄下)
logging.basicConfig(filename='serverless_error.log', level=logging.ERROR)
# 需要驗證的函式列表
PROTECTED_FUNCTIONS=['list_functions',
'add_function',
'save_function',
'edit_function',
'update_function',
'delete_function',
'show_stats',
'clear_stats'
]
# 登入管理功能
@app.route('/login', methods=['GET', 'POST'])
def login():
# GET 請求 : 顯示登入頁面
if request.method == 'GET':
return '''
<!DOCTYPE html>
<html>
<head><title>系統登入</title></head>
<body>
<h2>系統登入</h2>
<form method="post">
<input type="password" name="token" placeholder="請輸入密碼" required>
<button type="submit">登入</button>
</form>
</body>
</html>
'''
# POST 請求 : 處理登入請求
if request.is_json: # JSON 登入
token=request.json.get('token')
else: # 表單登入
token=request.form.get('token')
if token == SECRET_TOKEN: # 驗證登入密碼
# 將登入狀態儲存在瀏覽器的 session cookie 中 (以明碼方式儲存)
# Flask 會用金鑰對資料進行簽章 (非加密) 確保內容未被竄改
session['authenticated']=True
# 回應登入成功
if request.is_json:
return jsonify({'message': '登入成功'})
else:
return '<p>登入成功！<a href="/function/list_functions">查看函式列表</a></p>'
else: # 密碼錯誤 : 回應登入失敗訊息
if request.is_json:
return jsonify({'message': '登入失敗'}), 401
else:
return '<p>登入失敗！<a href="/login">重新登入</a></p>', 401
# 登出管理功能
@app.route('/logout')
def logout():
session.clear() # 清除伺服端 Flask session 字典中的所有鍵值對
return '<p>已登出！<a href="/login">重新登入</a></p>'
# 動態載入 & 執行函式模組 (支援 RESTful)
@app.route('/function/<func_name>', defaults={'subpath': ''}, methods=['GET', 'POST'])
@app.route('/function/<func_name>/<path:subpath>', methods=['GET', 'POST'])
def handle_function(func_name, subpath): # 傳入 subpath 支援 RESTful
# 1. 如果呼叫管理模組必須使用者已登入才行
if func_name in PROTECTED_FUNCTIONS and not check_auth():
return jsonify({'error': 'Authentication required', 'login_url': '/login'}), 401
# 2. 取得檔案路徑
func_path=os.path.join(FUNCTIONS_DIR, f'{func_name}.py')
if not os.path.isfile(func_path): # 模組檔案不存在 -> 回 404
return jsonify({'error': f'Function "{func_name}" not found'}), 404
try:
# 3. 動態載入模組 (絕對路徑)
spec=importlib.util.spec_from_file_location(func_name, func_path)
module=importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# 4. 檢查模組中有無 main() 函式 :
if not hasattr(module, 'main'): # 模組中無 main() 函式
return jsonify({'error': f'Module "{func_name}" has no main()'}), 400
# 5. 將 subpath 加入 request 中 (支援 RESTful)
request.view_args['subpath']=subpath
# 6. 記錄呼叫統計 (除了統計查詢自身避免無限循環)
if func_name not in PROTECTED_FUNCTIONS:
record_call(func_name)
# 7. 執行模組中的函式 (傳入模組可能需要的參數-但不一定會用到) :
result=module.main(request, config=config, protected=PROTECTED_FUNCTIONS)
# 8. 傳回函式執行結果
return result
except Exception as e:
logging.exception(f'Error in function {func_name}\n{e}') # 紀錄錯誤於日誌
return jsonify({'error': 'Function execution failed'}), 500
if __name__ == '__main__':
init_db() # 初始化資料庫
app.run(debug=True)
其中黃底色部分為呼叫紀錄功能所增加的程式碼, 主要就是新增 show_stats.py 與 clear_stats.py 這兩個函式模組, 它們也是列在被保護的模組, 不會顯示在 list_functions.py 呈現的模組列表 (無法限上編輯與刪除). 其次是新增了 init_db() 與 record_call() 這兩個函式, 當主程式執行時會先呼叫 init_db() 初始化 SQLite 資料庫 serverless.db, 建立一個 call_stats 資料表來紀錄函式模組名稱與被呼叫次數, 除了被保護模組外的其他模組每次被動態載入前都會先呼叫 record_call() 函式讓其呼叫次數增量 1. 注意, 主程式只記錄非管理模組被呼叫之累加次數.
另外, 本次改版也修改了動態載入模組時呼叫 module.main() 的傳入參數結構, 多了一個 protected 參數把管理模組名稱串列 PROTECTED_FUNCTIONS 傳給被載入之模組, 因為在 list_functions.py 與 show_stats.py 中列表中不需要顯示管理模組, 這時就需要用 protected 參數來判斷. 在函式模組中從 **kwargs 取出 protected 的語法如下 :
def main(request, **kwargs):
# 取得主程式傳遞之 protected 參數 (被保護之管理模組名稱)
protected=kwargs.get('protected', [])
這個 **kwargs 參數包含全部主程式所傳遞的參數字典, 不是每個函式模組都會用到, 會用到的就呼叫 kwargs.get() 自行取用.
2. 建立顯示呼叫統計的函式模組 show_stats.py :
此新增模組用來顯示各模組 (被保護模組除外) 的呼叫次數統計, 主要動作是連線 SQLite 資料庫 serverless.db 讀取其中的 call_stats 資料表, 然後用迴圈產生模組被呼叫次數表格之 HTML 字串後傳回 :
# show_stats.py
import sqlite3
DB_PATH='./serverless.db'
def main(request, **kwargs):
# 取得主程式傳遞之 protected 參數 (被保護之管理模組名稱)
protected=kwargs.get('protected', [])
# 連線資料庫
conn=sqlite3.connect(DB_PATH)
cursor=conn.cursor()
cursor.execute('SELECT func_name, call_count FROM call_stats ORDER BY func_name')
rows=cursor.fetchall()
conn.close()
# 產生回應表格
html='<h2>函式呼叫統計</h2>'
html += '<table border="1" cellpadding="6" cellspacing="0" style="border-collapse: collapse;">'
html += '<tr><th>函式名稱</th><th>呼叫次數</th></tr>'
for func_name, count in rows:
if func_name in protected: # 不顯示管理模組之被呼叫次數
continue
html += f'<tr><td>{func_name}</td><td>{count}</td></tr>'
html += '</table>'
html += '<br><a href="/function/clear_stats">清除統計資料</a> '
html += '<a href="/function/list_functions">返回函式列表</a>'
return html
雖然此模組是被動態載入到主程式 serverless.py 中執行, 主程式已經匯入 sqlite 模組也有定義 DB_PATH, 但 show_stats 本身是獨立的模組, 在執行其 main() 函式時, 裡面用到的外部資源 (像是 sqlite3 與資料庫路徑 DB_PATH) 都需要自己在這個模組內 import 並設定之, 因為它們是互不影響的不同模組空間.
其次, 此模組不顯示管理模組之輩呼叫次數, 故先用 kwargs.get() 取出主程式傳遞的 protected 參數以便在迴圈中用 continue 跳過它們.
3. 建立清除呼叫統計的函式模組 clear_stats.py :
此函式模組用來刪除 serverless.d 資料庫中的 call_stats 資料表中的全部紀錄, 重新統計函式呼叫次數 :
# clear_stats.py
import sqlite3
from flask import jsonify, session
DB_PATH='./serverless.db'
def main(request, **kwargs):
# 權限檢查確保只有登入用戶能清除呼叫統計
if not session.get('authenticated'):
return jsonify({'error': 'Authentication required'}), 401
# 清除 call_stats 資料表
try:
conn=sqlite3.connect(DB_PATH)
cursor=conn.cursor()
cursor.execute('DELETE FROM call_stats')
conn.commit()
conn.close()
return f'''
<p>已成功清除呼叫統計資料.</p>
<a href="/function/show_stats">返回呼叫統計列表</a>
'''
except Exception as e:
return jsonify({'error': f'清除呼叫統計失敗: {e}'}), 500
為了保險起見, 函式開頭會先驗證是否為已登入狀態, 成功清除後會顯示回呼叫統計列表的連結.
4. 修改顯示函式模組列表程式 list_functions.py :
模組 list_functions.py 作為本平台管理頁面的儀錶板, 所以我在函式模組列表上方添加一個 "呼叫統計" 超連結, 其餘內容不變 :
# list_functions.py
import os
def main(request, **kwargs):
# 取得主程式傳遞之 protected 參數 (被保護之管理模組名稱)
proteted=kwargs.get('protected', [])
# 取得 functions 目錄絕對路徑
functions_dir='./functions'
# 取得所有 .py 檔案（但排除 __init__.py）
try:
files=os.listdir(functions_dir)
py_files=[f[:-3] for f in files if f.endswith('.py') and f != '__init__.py']
py_files.sort() # 按字母順序排序
except FileNotFoundError:
return '<p>directory ./functions not found!</p>'
# 產生 HTML 碼
html = '<h2>函式列表</h2>'
html += '<table border="1" cellspacing="0" cellpadding="6" style="border-collapse: collapse;">'
html += '<tr><th>函式名稱</th><th>執行</th><th>編輯</th><th>刪除</th></tr>'
for func in py_files:
if func in protected: # 不顯示管理模組
continue
html += f'<tr>'
html += f'<td>{func}</td>'
html += f'<td><a href="/function/{func}">執行</a></td>'
html += f'<td><a href="/function/edit_function?module_name={func}">編輯</a></td>'
html += f'<td><a href="/function/delete_function?module_name={func}">刪除</a></td>'
html += f'</tr>'
html += '</table>'
html += '<br><a href="/function/add_function">新增函式</a> '
html += '<a href="/function/show_stats">呼叫統計</a> '
html += '<a href="/logout">登出</a>'
return html
此模組主要修改之處為黃底高亮部分, 增加了從 protected 參數中取得主程式傳遞的 PROTECT_FUNCTIONS 串列, 供迴圈中判斷是否為管理模組, 是的話就不顯示, 這樣就只要在主程式中維護一份 PROTECT_FUNCTIONS 串列即可; 其次是在底下增加一個呼叫 show_stats 模組的超連結前往顯示呼叫統計次數頁面.
由於更改了 serverless.py, 所以需重啟 serverless.service (手動刪除 serverless.db 則不需要重啟服務):
tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl restart serverless
瀏覽 list_functions 頁面 :
點各模組的 "執行" 超連結各一次後, 按 "呼叫統計" 超連結會顯示呼叫次數統計 :
按 "清除統計" 連結會清空呼叫次數統計資料表 call_stats :
按 "返回呼叫統計列表" 連結顯示為空 :
從呼叫統計即可明瞭各函式模組被呼叫之次數, 例如 LINE Bot 程式回應了多少次. 我將此新版 v3 的壓縮檔存放於 GitHub :
沒有留言 :
張貼留言