Heroku 上 Flask API 与 Dash 应用的部署与集成

from flask import Flask, request, jsonify, make_response
from flask_cors import CORS
import dash
from dash import dcc, html, Input, Output
import json
import os # 用于获取数据库连接字符串

# 1. 创建主 Flask 应用实例
app = Flask(__name__)
CORS(app) # 为主 Flask 应用启用 CORS

# 2. 将 Dash 应用集成到现有的 Flask 应用中
# 通过 server=app 参数，Dash 会使用我们已经创建的 Flask 应用实例
# url_base_pathname 可以指定 Dash 应用的根路径，例如 /dashboard/
dash_app = dash.Dash(__name__, server=app, url_base_pathname='/dashboard/')

# 3. 定义 Flask API 路由
# 这个路由现在属于主 Flask 应用
@app.route('/ingest', methods=['OPTIONS', 'POST'])
def handle_ingest():
    # 处理 CORS 预检请求
    if request.method == 'OPTIONS':
        response = make_response()
        response.headers.add('Access-Control-Allow-Origin', '*') # 生产环境请指定具体域名
        response.headers.add('Access-Control-Allow-Headers', 'Authorization, Content-Type')
        response.headers.add('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        return response

    # 认证逻辑
    token = request.headers.get('Authorization')
    # 客户端发送的是 'Bearer too_many_secrets'，所以这里也要匹配
    valid_tokens = ["Bearer too_many_secrets"] 

    if token in valid_tokens:
        data = request.json # 假设数据以 JSON 格式发送

        # --- 在这里执行数据验证和写入 PostgreSQL 数据库的逻辑 ---
        # 示例：连接到 Heroku Postgres 数据库并插入数据
        # import psycopg2
        # DATABASE_URL = os.environ.get('DATABASE_URL') # Heroku 会自动提供
        # try:
        #     conn = psycopg2.connect(DATABASE_URL, sslmode='require')
        #     cur = conn.cursor()
        #     # 示例：创建一个表并插入数据
        #     # cur.execute("CREATE TABLE IF NOT EXISTS sensor_data (id SERIAL PRIMARY KEY, sensor TEXT, value REAL, timestamp TIMESTAMPTZ DEFAULT NOW());")
        #     # cur.execute("INSERT INTO sensor_data (sensor, value) VALUES (%s, %s);", (data.get('sensor'), data.get('value')))
        #     conn.commit()
        #     cur.close()
        #     conn.close()
        #     print(f"Data ingested successfully: {data}")
        #     return jsonify({"message": "Data ingested successfully", "received_data": data}), 200
        # except Exception as e:
        #     print(f"Database error: {e}")
        #     return jsonify({"message": "Failed to ingest data due to database error"}), 500

        # 仅为演示，实际应写入数据库
        print(f"Success: Data ingested successfully: {data}")
        return jsonify({"message": "Data ingested successfully", "received_data": data}), 200
    else:
        print("Unauthorized user: Your token was Invalid")
        return jsonify({"message": "Unauthorized"}), 401

# 4. 定义 Dash 应用的布局和回调
# Dash 应用现在是主 Flask 应用的一个部分
dash_app.layout = html.Div(children=[
    html.H1(children='Heroku 集成应用'),
    html.P('欢迎来到 Dash 仪表板！'),
    dcc.Link('访问数据摄取 API 端点', href='/ingest', refresh=True), # 链接到 Flask API
    html.Div(id='output-message', style={'margin-top': '20px'})
])

# 示例 Dash 回调 (如果需要)
# @dash_app.callback(
#     Output('output-message', 'children'),
#     Input('url', 'pathname') # 需要 dcc.Location 组件才能获取 pathname
# )
# def display_page(pathname):
#     if pathname == '/dashboard/':
#         return html.Div("您正在查看 Dash 仪表板首页。")
#     elif pathname == '/ingest':
#         return html.Div("您已点击了 API 端点链接，但此页面本身不提供交互。")
#     return html.Div("未知页面")


# 5. 主程序入口
if __name__ == '__main__':
    # 在本地运行，Flask 应用将作为主服务器
    app.run(debug=True)

import requests

data = {
    "sensor": "temperature",
    "value": 25.5
}

# 假设 Heroku 应用的 URL 是 'https://my_app.herokuapp.com/'
# API 端点现在是 '/ingest'
api_endpoint = 'https://my_app.herokuapp.com/ingest'  
token = 'too_many_secrets' # 客户端的原始 token

headers = {'Authorization': f'Bearer {token}'} # 按照约定发送 Bearer token

response = requests.post(api_endpoint, json=data, headers=headers, verify=True)

if response.status_code == 200:
    print("Data sent successfully")
    print(f"Response: {response.json()}")
else:
    print(f"Failed to send data. Status code: {response.status_code}")
    print(f'Response content: {response.content.decode()}') # 解码以便阅读

web: gunicorn app:app