### Python 实现数据库查询、数据可视化及数据同步至接口 #### 数据库查询 Python 可以通过 `pymysql` 或者 `sqlalchemy` 库来连接 MySQL 数据库并执行 SQL 查询语句。以下是使用 `pymysql` 进行数据库查询的代码示例： ```python import pymysql def query_data(host, user, password, db, sql_query): connection = None try: # 建立数据库连接 connection = pymysql.connect( host=host, user=user, password=password, database=db, cursorclass=pymysql.cursors.DictCursor ) with connection.cursor() as cursor: # 执行SQL查询 cursor.execute(sql_query) result = cursor.fetchall() return result except Exception as e: print(f"Error occurred while querying the database: {e}") finally: if connection: connection.close() # 调用函数 data = query_data('localhost', 'root', 'password', 'pyt', 'SELECT * FROM your_table LIMIT 10') print(data) ``` 以上代码实现了从指定数据库中提取前 10 条记录的功能[^1]。 --- #### 数据可视化 对于数据可视化，可以选择多种工具，如 Plotly 和 Altair。以下分别介绍两种方式实现简单的图表绘制。 ##### 使用 Plotly 绘制折线图 Plotly 是一种支持交互式可视化的工具，适合用于动态展示数据变化趋势。 ```python import pandas as pd import plotly.express as px df = pd.DataFrame(data) # 将查询结果转化为 DataFrame fig = px.line(df, x='column_x', y='column_y', title="Line Chart Example") fig.show() ``` 此处假设返回的结果集中存在名为 `column_x` 和 `column_y` 的字段作为横纵坐标轴[^1]。 ##### 使用 Altair 创建散点图 Altair 提供了一种简洁的方式构建静态或互动型图表。 ```python import altair as alt chart = alt.Chart(df).mark_circle().encode( x='column_x', y='column_y', color='category_column' ).interactive() chart.display() ``` 这里同样假定有一个分类变量 `category_column` 存在于原始数据集之中[^2]。 --- #### 数据选择与同步到接口 当用户希望手动挑选部分数据并通过 RESTful API 发送到远程服务端时，可借助 Flask 构建本地 Web 接口完成这一目标。 ##### 步骤一：创建前端页面让用户勾选所需项 HTML 表单允许提交多个选项给后端处理器。 ```html <form action="/submit_selections" method="POST"> {% for row in data %} <input type="checkbox" name="selected_rows" value="{{row.id}}"> {{row.name}}<br> {% endfor %} <button type="submit">Submit</button> </form> ``` ##### 步骤二：接收 POST 请求并将选定的数据发送出去 Flask 控制器负责解析请求参数再调用外部 HTTP 客户端推送消息。 ```python from flask import Flask, request import requests app = Flask(__name__) @app.route('/submit_selections', methods=['POST']) def submit_selections(): selected_ids = request.form.getlist('selected_rows') # 获取所选ID列表 filtered_data = [item for item in data if str(item['id']) in selected_ids] url = "http://example.com/api/endpoint" headers = {'Content-Type': 'application/json'} response = requests.post(url, json=filtered_data, headers=headers) return f"Data sent successfully! Status code: {response.status_code}" if __name__ == '__main__': app.run(debug=True) ``` 上述片段展示了如何捕获来自用户的输入，并将其转发至第三方 URL 地址[^3]。 ---