timmy / 通用 Flask API 模板

#!/usr/bin/env python import os import sys from urllib.parse import quote_plus from datetime import datetime from dotenv import load_dotenv from flask import Flask, request, jsonify, g import records import pandas as pd from loguru import logger # 載入 .env 檔案中的環境變數 load_dotenv() # 應用程式與 API 設定參數 SECRET_KEY = os.getenv("SECRET_KEY", "default_secret") API_KEY = os.getenv("API_KEY", "your_generic_api_key_here") LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") # 設定 loguru 日誌工具，將日誌輸出到標準輸出 logger.remove() logger.add( sys.stdout, level=LOG_LEVEL, format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level}</level> | <cyan>{message}</cyan>", ) # ------------------ 資料庫連線模組 ------------------ class BaseDatabase: def __init__(self, url): self.url = url logger.debug(f"資料庫 URL: {self.url}") try: self.db = records.Database(self.url) logger.info("資料庫連線初始化成功。") except Exception as e: logger.error(f"資料庫連線初始化失敗: {e}") raise def query(self, sql): try: logger.debug(f"執行 SQL 查詢: {sql}") self.result = self.db.query(sql, fetchall=True) logger.info("查詢執行成功。") except Exception as e: logger.error(f"查詢執行發生錯誤: {e}") self.result = None def to_dataframe(self): if self.result: try: data = self.result.dataset.dict logger.debug(f"查詢結果轉換為 DataFrame: {data}") return pd.DataFrame.from_dict(data) except Exception as e: logger.error(f"轉換為 DataFrame 失敗: {e}") return pd.DataFrame() else: logger.warning("查詢結果為空。") return pd.DataFrame() def close(self): try: self.db.close() logger.info("資料庫連線已成功關閉。") except Exception as e: logger.error(f"關閉資料庫連線失敗: {e}") def test_connection(self): try: self.query("SELECT 1+1 AS result") if self.result: for row in self.result: row_dict = row.as_dict() return row_dict.get("result") else: logger.warning("測試連線查詢無回傳結果。") return None except Exception as e: logger.error(f"測試連線時發生錯誤: {e}") return None class SqlSrv(BaseDatabase): def __init__(self, server, username, password, db_name): if None in [server, username, password, db_name]: raise ValueError("資料庫連線參數不足。") password_encoded = quote_plus(password) url = f"mssql+pymssql://{username}:{password_encoded}@{server}:1433/{db_name}?tds_version=7.0" super().__init__(url) class SQLite(BaseDatabase): def __init__(self, db_path): url = f"sqlite:///{db_path}" super().__init__(url) class MySQL(BaseDatabase): def __init__(self, server, username, password, db_name): url = f"mysql+pymysql://{username}:{quote_plus(password)}@{server}/{db_name}" super().__init__(url) def get_db(): """ 從環境變數中讀取資料庫連線參數， 建立 SQL Server 連線（可根據需求改用 SQLite 或 MySQL）。 """ DB_SERVER = os.getenv("DB_SERVER") DB_USERNAME = os.getenv("DB_USERNAME") DB_PASSWORD = os.getenv("DB_PASSWORD") DB_NAME = os.getenv("DB_NAME") return SqlSrv(DB_SERVER, DB_USERNAME, DB_PASSWORD, DB_NAME) # ------------------ API 工具與路由設定 ------------------ def validate_api_key(): """ 檢查請求中的 API 金鑰是否正確。 """ key = request.headers.get("Authorization") or request.headers.get("X-API-KEY") if not key or (key != f"Bearer {API_KEY}" and key != API_KEY): return False return True def create_app(): """ 建立並配置 Flask 應用程式， 包含資料庫連線初始化、路由註冊等設定。 """ app = Flask(__name__) app.config['SECRET_KEY'] = SECRET_KEY @app.before_request def before_request(): """為每個請求初始化資料庫連線。""" g.db = get_db() @app.teardown_request def teardown_request(exception=None): """在請求結束時關閉資料庫連線。""" db = getattr(g, 'db', None) if db is not None: db.close() @app.route('/data', methods=['GET']) def get_data(): """ 資料查詢 API： - 驗證 API 金鑰 - 解析請求參數（start 與 end） - 執行 SQL 查詢並回傳 JSON 格式的查詢結果 """ if not validate_api_key(): return jsonify({"error": "Unauthorized"}), 401 start_time = request.args.get('start') end_time = request.args.get('end') if not start_time or not end_time: return jsonify({"error": "Please provide 'start' and 'end' parameters"}), 400 try: start_dt = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S") end_dt = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S") except ValueError: return jsonify({"error": "Invalid date format. Use YYYY-MM-DD HH:MM:SS"}), 400 # 示範查詢語法，請根據實際需求修改資料表名稱與欄位 query = f""" SELECT * FROM my_table WHERE created_at >= '{start_dt}' AND created_at <= '{end_dt}' ORDER BY created_at ASC """ g.db.query(query) df = g.db.to_dataframe() if df.empty: return jsonify([]) return jsonify(df.to_dict(orient="records")) @app.route('/health', methods=['GET']) def health_check(): """健康檢查 API，回傳伺服器狀態。""" return jsonify({"status": "ok"}) return app # ------------------ 主程式執行進入點 ------------------ if __name__ == '__main__': from waitress import serve app = create_app() print("Starting generic server with Waitress...") serve(app, host='0.0.0.0', port=5000)

#!/bin/bash # 測試 API /data 端點的 curl 測試 script # Usage: ./test_api.sh '2025-01-01 00:00:00' '2025-01-02 00:00:00' API_URL="http://127.0.0.1:5000/data" API_KEY="your_generic_api_key_here" # 請替換成正確的 API 金鑰 if [ "$#" -lt 2 ]; then echo "Usage: $0 <start_time> <end_time>" echo "Example: $0 '2025-01-01 00:00:00' '2025-01-02 00:00:00'" exit 1 fi START_TIME="$1" END_TIME="$2" # 將時間中的空格轉換成 URL 可讀格式 ENCODED_START=$(echo "$START_TIME" | sed 's/ /%20/g') ENCODED_END=$(echo "$END_TIME" | sed 's/ /%20/g') curl -H "Authorization: Bearer $API_KEY" \ "$API_URL?start=$ENCODED_START&end=$ENCODED_END"