J'aime 1
Fork 0

Dernière activité 9 months ago

提供一個整合資料庫連線、API 金鑰驗證與基本路由的 Flask 應用程式範本，適合作為 Web API 開發起點。

api database flask python rest template

Révision df402c41896a2dec8a51811e3cab9a29e0ce8df7

README.md Brut

專案簡介

本專案展示如何整合環境變數管理、日誌系統、資料庫連線封裝與 API 設計，並透過 Flask 框架提供 RESTful 服務。專案強調資源管理與安全驗證的流程，適合用於建立穩定且安全的後端服務。

初始化與環境設定

  1. 環境變數讀取
    程式啟動時會使用 dotenv 載入 .env 檔案中的設定，包含 API 金鑰、資料庫連線參數、日誌等級及密鑰等。

  2. 日誌工具設定
    利用 loguru 設定日誌輸出，記錄連線狀態與錯誤資訊，方便追蹤與除錯。

資料庫連線模組

  1. BaseDatabase 類別
    封裝與資料庫互動的基本流程，包含建立連線、執行 SQL 查詢、將查詢結果轉換為 DataFrame 以及正確關閉連線。

  2. 子類別擴展
    根據不同資料庫（如 SQL Server、SQLite、MySQL），定義各自的子類別，依資料庫連線格式產生合適的連線 URL，並調用 BaseDatabase 功能。

資料庫連線管理

  • get_db 函式
    從環境變數中讀取 SQL Server 的連線參數，建立資料庫連線實例，確保每次請求都能根據設定取得正確的連線。

Flask 應用程式建立與配置

  1. 建立 Flask 應用
    透過 create_app() 函式建立並配置 Flask 應用，並設定必要參數（例如 SECRET_KEY）。

  2. 生命週期鉤子

    • before_request：每個 HTTP 請求前，請透過 get_db() 初始化資料庫連線，並將連線存放於全域變數 g 中。
    • teardown_request：請求結束後，不論是否發生錯誤，都會關閉資料庫連線，以確保資源正確釋放。

API 路由與邏輯處理

  1. API 金鑰驗證
    請在進入 API 邏輯前，透過 validate_api_key() 驗證請求是否包含正確的 API 金鑰，以保護 API 存取權限。

  2. 資料查詢 API (/data)

    • 接收 start 與 end 兩個時間參數，並驗證其格式是否正確。
    • 根據提供的時間區間構造 SQL 查詢，從資料表中抓取符合條件的資料。
    • 查詢結果將轉換為 Pandas DataFrame，最終以 JSON 格式回傳給請求端。

  3. 健康檢查 API (/health)
    提供一個簡單的端點來回報服務狀態，常用於監控或自動化健康檢查。

主程式執行

  • 使用 Waitress 啟動服務
    主程式區塊中將建立好的 Flask 應用交由 Waitress 伺服器運行，使服務能在指定主機與埠口上穩定提供 HTTP 服務。

整體而言，此專案展示了如何整合多項技術與流程，達成穩定、安全的後端服務。請依據實際需求進行修改與擴充，如有任何問題，歡迎提出討論。

template.py Brut
1#!/usr/bin/env python
2
3import os
4import sys
5from urllib.parse import quote_plus
6from datetime import datetime
7
8from dotenv import load_dotenv
9from flask import Flask, request, jsonify, g
10import records
11import pandas as pd
12from loguru import logger
13
14# 載入 .env 檔案中的環境變數
15load_dotenv()
16
17# 應用程式與 API 設定參數
18SECRET_KEY = os.getenv("SECRET_KEY", "default_secret")
19API_KEY = os.getenv("API_KEY", "your_generic_api_key_here")
20LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
21
22# 設定 loguru 日誌工具，將日誌輸出到標準輸出
23logger.remove()
24logger.add(
25 sys.stdout,
26 level=LOG_LEVEL,
27 format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level}</level> | <cyan>{message}</cyan>",
28)
29
30# ------------------ 資料庫連線模組 ------------------
31
32class BaseDatabase:
33 def __init__(self, url):
34 self.url = url
35 logger.debug(f"資料庫 URL: {self.url}")
36 try:
37 self.db = records.Database(self.url)
38 logger.info("資料庫連線初始化成功。")
39 except Exception as e:
40 logger.error(f"資料庫連線初始化失敗: {e}")
41 raise
42
43 def query(self, sql):
44 try:
45 logger.debug(f"執行 SQL 查詢: {sql}")
46 self.result = self.db.query(sql, fetchall=True)
47 logger.info("查詢執行成功。")
48 except Exception as e:
49 logger.error(f"查詢執行發生錯誤: {e}")
50 self.result = None
51
52 def to_dataframe(self):
53 if self.result:
54 try:
55 data = self.result.dataset.dict
56 logger.debug(f"查詢結果轉換為 DataFrame: {data}")
57 return pd.DataFrame.from_dict(data)
58 except Exception as e:
59 logger.error(f"轉換為 DataFrame 失敗: {e}")
60 return pd.DataFrame()
61 else:
62 logger.warning("查詢結果為空。")
63 return pd.DataFrame()
64
65 def close(self):
66 try:
67 self.db.close()
68 logger.info("資料庫連線已成功關閉。")
69 except Exception as e:
70 logger.error(f"關閉資料庫連線失敗: {e}")
71
72 def test_connection(self):
73 try:
74 self.query("SELECT 1+1 AS result")
75 if self.result:
76 for row in self.result:
77 row_dict = row.as_dict()
78 return row_dict.get("result")
79 else:
80 logger.warning("測試連線查詢無回傳結果。")
81 return None
82 except Exception as e:
83 logger.error(f"測試連線時發生錯誤: {e}")
84 return None
85
86class SqlSrv(BaseDatabase):
87 def __init__(self, server, username, password, db_name):
88 if None in [server, username, password, db_name]:
89 raise ValueError("資料庫連線參數不足。")
90 password_encoded = quote_plus(password)
91 url = f"mssql+pymssql://{username}:{password_encoded}@{server}:1433/{db_name}?tds_version=7.0"
92 super().__init__(url)
93
94class SQLite(BaseDatabase):
95 def __init__(self, db_path):
96 url = f"sqlite:///{db_path}"
97 super().__init__(url)
98
99class MySQL(BaseDatabase):
100 def __init__(self, server, username, password, db_name):
101 url = f"mysql+pymysql://{username}:{quote_plus(password)}@{server}/{db_name}"
102 super().__init__(url)
103
104def get_db():
105 """
106 從環境變數中讀取資料庫連線參數，
107 建立 SQL Server 連線（可根據需求改用 SQLite 或 MySQL）。
108 """
109 DB_SERVER = os.getenv("DB_SERVER")
110 DB_USERNAME = os.getenv("DB_USERNAME")
111 DB_PASSWORD = os.getenv("DB_PASSWORD")
112 DB_NAME = os.getenv("DB_NAME")
113 return SqlSrv(DB_SERVER, DB_USERNAME, DB_PASSWORD, DB_NAME)
114
115# ------------------ API 工具與路由設定 ------------------
116
117def validate_api_key():
118 """
119 檢查請求中的 API 金鑰是否正確。
120 """
121 key = request.headers.get("Authorization") or request.headers.get("X-API-KEY")
122 if not key or (key != f"Bearer {API_KEY}" and key != API_KEY):
123 return False
124 return True
125
126def create_app():
127 """
128 建立並配置 Flask 應用程式，
129 包含資料庫連線初始化、路由註冊等設定。
130 """
131 app = Flask(__name__)
132 app.config['SECRET_KEY'] = SECRET_KEY
133
134 @app.before_request
135 def before_request():
136 """為每個請求初始化資料庫連線。"""
137 g.db = get_db()
138
139 @app.teardown_request
140 def teardown_request(exception=None):
141 """在請求結束時關閉資料庫連線。"""
142 db = getattr(g, 'db', None)
143 if db is not None:
144 db.close()
145
146 @app.route('/data', methods=['GET'])
147 def get_data():
148 """
149 資料查詢 API：
150 - 驗證 API 金鑰
151 - 解析請求參數（start 與 end）
152 - 執行 SQL 查詢並回傳 JSON 格式的查詢結果
153 """
154 if not validate_api_key():
155 return jsonify({"error": "Unauthorized"}), 401
156
157 start_time = request.args.get('start')
158 end_time = request.args.get('end')
159
160 if not start_time or not end_time:
161 return jsonify({"error": "Please provide 'start' and 'end' parameters"}), 400
162
163 try:
164 start_dt = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
165 end_dt = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
166 except ValueError:
167 return jsonify({"error": "Invalid date format. Use YYYY-MM-DD HH:MM:SS"}), 400
168
169 # 示範查詢語法，請根據實際需求修改資料表名稱與欄位
170 query = f"""
171 SELECT *
172 FROM my_table
173 WHERE created_at >= '{start_dt}'
174 AND created_at <= '{end_dt}'
175 ORDER BY created_at ASC
176 """
177 g.db.query(query)
178 df = g.db.to_dataframe()
179
180 if df.empty:
181 return jsonify([])
182
183 return jsonify(df.to_dict(orient="records"))
184
185 @app.route('/health', methods=['GET'])
186 def health_check():
187 """健康檢查 API，回傳伺服器狀態。"""
188 return jsonify({"status": "ok"})
189
190 return app
191
192# ------------------ 主程式執行進入點 ------------------
193
194if __name__ == '__main__':
195 from waitress import serve
196 app = create_app()
197 print("Starting generic server with Waitress...")
198 serve(app, host='0.0.0.0', port=5000)
199
test_api.sh Brut
1#!/bin/bash
2
3# 測試 API /data 端點的 curl 測試 script
4# Usage: ./test_api.sh '2025-01-01 00:00:00' '2025-01-02 00:00:00'
5
6API_URL="http://127.0.0.1:5000/data"
7API_KEY="your_generic_api_key_here" # 請替換成正確的 API 金鑰
8
9if [ "$#" -lt 2 ]; then
10 echo "Usage: $0 <start_time> <end_time>"
11 echo "Example: $0 '2025-01-01 00:00:00' '2025-01-02 00:00:00'"
12 exit 1
13fi
14
15START_TIME="$1"
16END_TIME="$2"
17
18# 將時間中的空格轉換成 URL 可讀格式
19ENCODED_START=$(echo "$START_TIME" | sed 's/ /%20/g')
20ENCODED_END=$(echo "$END_TIME" | sed 's/ /%20/g')
21
22curl -H "Authorization: Bearer $API_KEY" \
23 "$API_URL?start=$ENCODED_START&end=$ENCODED_END"
24

Propulsé par Opengist ⋅ Load: 110ms⋅

Français