Rewizja 通用 Flask API 模板

timmy zrewidował ten Gist 9 months ago. Przejdź do rewizji

1 file changed, 68 insertions

README.md(stworzono plik)

		@@ -0,0 +1,68 @@
1	+	# 專案簡介
2	+
3	+	本專案展示如何整合環境變數管理、日誌系統、資料庫連線封裝與 API 設計，並透過 Flask 框架提供 RESTful 服務。專案強調資源管理與安全驗證的流程，適合用於建立穩定且安全的後端服務。
4	+
5	+	---
6	+
7	+	## 初始化與環境設定
8	+
9	+	1. 環境變數讀取
10	+	程式啟動時會使用 `dotenv` 載入 .env 檔案中的設定，包含 API 金鑰、資料庫連線參數、日誌等級及密鑰等。
11	+
12	+	2. 日誌工具設定
13	+	利用 loguru 設定日誌輸出，記錄連線狀態與錯誤資訊，方便追蹤與除錯。
14	+
15	+	---
16	+
17	+	## 資料庫連線模組
18	+
19	+	1. BaseDatabase 類別
20	+	封裝與資料庫互動的基本流程，包含建立連線、執行 SQL 查詢、將查詢結果轉換為 DataFrame 以及正確關閉連線。
21	+
22	+	2. 子類別擴展
23	+	根據不同資料庫（如 SQL Server、SQLite、MySQL），定義各自的子類別，依資料庫連線格式產生合適的連線 URL，並調用 BaseDatabase 功能。
24	+
25	+	---
26	+
27	+	## 資料庫連線管理
28	+
29	+	- get_db 函式
30	+	從環境變數中讀取 SQL Server 的連線參數，建立資料庫連線實例，確保每次請求都能根據設定取得正確的連線。
31	+
32	+	---
33	+
34	+	## Flask 應用程式建立與配置
35	+
36	+	1. 建立 Flask 應用
37	+	透過 `create_app()` 函式建立並配置 Flask 應用，並設定必要參數（例如 SECRET_KEY）。
38	+
39	+	2. 生命週期鉤子
40	+	- before_request：每個 HTTP 請求前，請透過 `get_db()` 初始化資料庫連線，並將連線存放於全域變數 `g` 中。
41	+	- teardown_request：請求結束後，不論是否發生錯誤，都會關閉資料庫連線，以確保資源正確釋放。
42	+
43	+	---
44	+
45	+	## API 路由與邏輯處理
46	+
47	+	1. API 金鑰驗證
48	+	請在進入 API 邏輯前，透過 `validate_api_key()` 驗證請求是否包含正確的 API 金鑰，以保護 API 存取權限。
49	+
50	+	2. 資料查詢 API (`/data`)
51	+
52	+	- 接收 `start` 與 `end` 兩個時間參數，並驗證其格式是否正確。
53	+	- 根據提供的時間區間構造 SQL 查詢，從資料表中抓取符合條件的資料。
54	+	- 查詢結果將轉換為 Pandas DataFrame，最終以 JSON 格式回傳給請求端。
55	+
56	+	3. 健康檢查 API (`/health`)
57	+	提供一個簡單的端點來回報服務狀態，常用於監控或自動化健康檢查。
58	+
59	+	---
60	+
61	+	## 主程式執行
62	+
63	+	- 使用 Waitress 啟動服務
64	+	主程式區塊中將建立好的 Flask 應用交由 Waitress 伺服器運行，使服務能在指定主機與埠口上穩定提供 HTTP 服務。
65	+
66	+	---
67	+
68	+	整體而言，此專案展示了如何整合多項技術與流程，達成穩定、安全的後端服務。請依據實際需求進行修改與擴充，如有任何問題，歡迎提出討論。

timmy zrewidował ten Gist 9 months ago. Przejdź do rewizji

2 files changed, 221 insertions

template.py(stworzono plik)

		@@ -0,0 +1,198 @@
1	+	#!/usr/bin/env python
2	+
3	+	import os
4	+	import sys
5	+	from urllib.parse import quote_plus
6	+	from datetime import datetime
7	+
8	+	from dotenv import load_dotenv
9	+	from flask import Flask, request, jsonify, g
10	+	import records
11	+	import pandas as pd
12	+	from loguru import logger
13	+
14	+	# 載入 .env 檔案中的環境變數
15	+	load_dotenv()
16	+
17	+	# 應用程式與 API 設定參數
18	+	SECRET_KEY = os.getenv("SECRET_KEY", "default_secret")
19	+	API_KEY = os.getenv("API_KEY", "your_generic_api_key_here")
20	+	LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
21	+
22	+	# 設定 loguru 日誌工具，將日誌輸出到標準輸出
23	+	logger.remove()
24	+	logger.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	+
32	+	class 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	+
86	+	class 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	+
94	+	class SQLite(BaseDatabase):
95	+	def __init__(self, db_path):
96	+	url = f"sqlite:///{db_path}"
97	+	super().__init__(url)
98	+
99	+	class 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	+
104	+	def 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	+
117	+	def 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	+
126	+	def 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	+
194	+	if __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)

test_api.sh(stworzono plik)

		@@ -0,0 +1,23 @@
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	+
6	+	API_URL="http://127.0.0.1:5000/data"
7	+	API_KEY="your_generic_api_key_here" # 請替換成正確的 API 金鑰
8	+
9	+	if [ "$#" -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
13	+	fi
14	+
15	+	START_TIME="$1"
16	+	END_TIME="$2"
17	+
18	+	# 將時間中的空格轉換成 URL 可讀格式
19	+	ENCODED_START=$(echo "$START_TIME" \| sed 's/ /%20/g')
20	+	ENCODED_END=$(echo "$END_TIME" \| sed 's/ /%20/g')
21	+
22	+	curl -H "Authorization: Bearer $API_KEY" \
23	+	"$API_URL?start=$ENCODED_START&end=$ENCODED_END"

Nowsze Starsze