☰
用 Python + Flask 写一个 MySQL 查询接口,返回 JSON 数据(附 TaoToken 统一 Key 配置)
2026/10/12 10:16:08 网站建设 项目流程

1. 从脚本到接口:Flask + MySQL 查询返回 JSON 到底解决什么问题

很多人第一次写 Python 查 MySQL,都是像下面这样:连上数据库、执行一条select * from user、把结果print出来,脚本一跑,控制台刷出一堆元组,任务就算完成了。这种写法自己看数据没问题,但一旦别人(前端、另一个服务、AI 工具)想拿这份数据,就抓瞎了——总不能让人家连你的数据库账号密码吧。

所以要把「查数据」这件事包装成一个 HTTP 接口:别人发一个请求,你返回一段 JSON,数据格式统一、跨语言通用。Flask 就是干这个最轻量的选择,几十行代码就能跑起来;MySQL 负责存数据;JSON 负责把查询结果序列化成标准格式。这套组合特别适合中小项目、内部工具、数据看板的后端,也适合你后面把接口接到 AI 工具链里做数据问答。

这篇文章我会带你从零走一遍:建表、写 Flask 路由、连接 MySQL、把结果转成 JSON、用 curl 验证,最后再把接口调用里的鉴权和 Base URL 统一收敛到 TaoToken,这样以后你接别的 AI 工具时,Key 和地址不用到处改。适合刚学 Flask 的同学,也适合手上有老脚本想改成接口的开发者。核心检索词就是 python、Flask、MySQL、接口、JSON,下面每一步都给可复制的代码。

先说清楚整体链路:客户端(curl / Postman / 前端)发 HTTP 请求 → Flask 路由接收参数 → 用数据库驱动连 MySQL 执行 SQL → 拿到结果集 → 逐行转成字典 →jsonify或json.dumps返回。中间最容易踩的坑有三个:一是数据库连接没关导致连接数爆掉;二是datetime类型不能直接 JSON 序列化;三是 SQL 拼接导致注入。这三个我都会在对应章节讲透。

2. 环境准备与 TaoToken 统一 Key 前置配置

在写代码之前,先把依赖和「统一入口」这两件事办了。依赖很简单,一个requirements.txt搞定:

Flask==3.0.3 PyMySQL==1.1.1

这里我特意用PyMySQL而不是老教程里的MySQLdb。原因很实际:MySQLdb(也就是mysqlclient)在 Windows 上编译经常报错,装起来费劲;PyMySQL是纯 Python 实现,pip install直接过,API 还基本兼容。你原来的脚本里import MySQLdb,换成import pymysql后,connect()的参数几乎不用改,后面我会给对照。

装依赖:

pip install -r requirements.txt

接下来是这篇比较特别的部分——把接口调用的鉴权和 Base URL 统一到 TaoToken。你可能会问:我一个查 MySQL 的接口,跟 AI 工具有啥关系?关系在于,很多人的项目里不止一个外部服务,今天接个模型对话,明天接个代码补全,后天又来个 Agent,每个服务的 Key 和地址都散落在不同文件里,改一次要翻半天。TaoToken 提供统一的 API 入口,把模型调用、Coding Plan、API Keys 管理这些收敛到一处,Base URL 固定为https://taotoken.net/api,你只需要维护一个 Key。

具体操作:先到控制台创建 API Key。打开 https://taotoken.net/console ,登录后在 API Keys 页面新建一个 Key,复制出来保存好(只显示一次)。这个 Key 就是你后面所有工具共用的凭证。

然后确认你的 Base URL。所有走 TaoToken 的请求,地址前缀都是:

https://taotoken.net/api

比如模型对话的调用、Coding Plan 的接入,都基于这个 Base URL 拼路径。文档在 https://taotoken.net/doc ,里面有各端点的具体路径和参数说明,建议收藏。

为什么要在写 Flask 接口的阶段就提这个?因为一个健康的习惯是:外部依赖的地址和密钥,永远从环境变量或统一配置读,不要硬编码在业务代码里。你的 Flask 接口连 MySQL 的密码不该写死,同理,你调 AI 服务的 Key 也不该写死。下面我给一个.env风格的配置约定,Flask 和 AI 调用共用:

# .env 示例,实际用 os.environ 读取 DB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=root DB_PASSWORD=your_db_password DB_NAME=zlnewschema TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_API_KEY=sk-你的Key

这样你的 Flask 接口负责查数据,AI 相关调用负责理解数据,两边共享同一套配置加载逻辑,迁移环境时只改.env,代码一行不动。如果你后面要做长期编码或 Agent 类项目,可以了解下 Coding Plan:https://taotoken.net/coding-plan ,它把编码场景的额度和管理也整合进来了。

3. 可复制配置:建表 SQL、Flask 路由与 JSON 序列化

这一节是核心,代码全部可复制。先建库建表,再灌测试数据。

CREATE DATABASE IF NOT EXISTS zlnewschema DEFAULT CHARSET utf8mb4; USE zlnewschema; CREATE TABLE IF NOT EXISTS user ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(64) NOT NULL, email VARCHAR(128) NOT NULL, password VARCHAR(128) NOT NULL, createtime DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; INSERT INTO user (name, email, password, createtime) VALUES ('张亮', 'zhangliang@example.com', 'pw_zhang', '2024-05-01 10:20:30'), ('李雷', 'lilei@example.com', 'pw_li', '2024-05-02 11:00:00'), ('韩梅梅', 'hanmeimei@example.com', 'pw_han', '2024-05-03 09:15:45');

注意createtime是DATETIME类型,这就是后面报TypeError的根源,先埋个伏笔。

接着写 Flask 应用。我把它拆成「配置加载 + 连接管理 + 路由」三块,方便你对照自己的老脚本改。

# app.py import os import json from datetime import datetime from flask import Flask, request, jsonify import pymysql app = Flask(__name__) DB_CONFIG = { "host": os.environ.get("DB_HOST", "127.0.0.1"), "port": int(os.environ.get("DB_PORT", 3306)), "user": os.environ.get("DB_USER", "root"), "password": os.environ.get("DB_PASSWORD", "your_db_password"), "database": os.environ.get("DB_NAME", "zlnewschema"), "charset": "utf8mb4", "cursorclass": pymysql.cursors.DictCursor, } def get_conn(): return pymysql.connect(**DB_CONFIG) def row_to_dict(row): """把一行数据转成可 JSON 序列化的字典""" result = {} for key, value in row.items(): if isinstance(value, datetime): result[key] = value.strftime("%Y-%m-%d %H:%M:%S") else: result[key] = value return result @app.route("/api/users", methods=["GET"]) def list_users(): name = request.args.get("name") conn = get_conn() try: with conn.cursor() as cursor: if name: sql = "SELECT id, name, email, createtime FROM user WHERE name = %s" cursor.execute(sql, (name,)) else: sql = "SELECT id, name, email, createtime FROM user" cursor.execute(sql) rows = cursor.fetchall() data = [row_to_dict(r) for r in rows] return jsonify({"code": 0, "msg": "ok", "data": data}) finally: conn.close() if __name__ == "__main__": app.run(host="0.0.0.0", port=5590, debug=True)

几个关键改动,对照你原来的脚本看:

第一,cursorclass=pymysql.cursors.DictCursor。这一行让查询结果直接是字典({'name': '张亮', ...}),而不是元组。你原来用row[0]、row[1]按位置取值,字段顺序一变就全乱;用字典游标后按字段名取值,稳得多。

第二,row_to_dict里对datetime做了strftime转换。这就是你之前遇到TypeError: Object of type 'datetime' is not JSON serializable的正解——不是把整行拼成字符串,而是逐字段判断类型再转换,这样返回的仍是结构化 JSON,不是一坨字符串。

第三,SQL 用%s占位符 + 参数元组(name,),而不是"where name = '%s'" % name。后者是字符串拼接,用户传个' OR '1'='1就能拖库,必须换成参数化查询。

第四,用jsonify而不是json.dumps。jsonify会自动设置Content-Type: application/json,还会处理中文编码,比手动dumps再Response省事。

如果你确实想保留json.dumps的写法(比如要控制indent),可以这样:

from flask import Response return Response( json.dumps({"code": 0, "data": data}, ensure_ascii=False, indent=2), mimetype="application/json" )

ensure_ascii=False保证中文不被转成\uXXXX,mimetype保证响应头正确。

4. 验证请求:curl 与 Postman 实测返回 JSON

代码写完,跑起来验证。先启动服务:

python app.py

看到Running on http://0.0.0.0:5590就说明起来了。然后开另一个终端,用 curl 测。

查全部:

curl -s http://127.0.0.1:5590/api/users

预期返回:

{ "code": 0, "msg": "ok", "data": [ {"id": 1, "name": "张亮", "email": "zhangliang@example.com", "createtime": "2024-05-01 10:20:30"}, {"id": 2, "name": "李雷", "email": "lilei@example.com", "createtime": "2024-05-02 11:00:00"}, {"id": 3, "name": "韩梅梅", "email": "hanmeimei@example.com", "createtime": "2024-05-03 09:15:45"} ] }

按条件筛选,查「张亮」:

curl -s "http://127.0.0.1:5590/api/users?name=张亮"

预期只返回一条:

{ "code": 0, "msg": "ok", "data": [ {"id": 1, "name": "张亮", "email": "zhangliang@example.com", "createtime": "2024-05-01 10:20:30"} ] }

注意 URL 里的中文,curl 一般能自动处理;如果遇到编码问题,可以用--data-urlencode配合 GET 的-G:

curl -s -G "http://127.0.0.1:5590/api/users" --data-urlencode "name=张亮"

用 Postman 的话,方法选 GET,URL 填http://127.0.0.1:5590/api/users,在 Params 里加一个name=张亮,点 Send,Body 选 Pretty + JSON,就能看到和上面一样的结构。你原来用 POST +request.json.get('username')也行,但查询类操作语义上更推荐 GET,参数放 query string,缓存和调试都更友好。如果坚持 POST,把路由改成methods=["POST"],然后:

payload = request.get_json(silent=True) or {} name = payload.get("username")

这里silent=True很关键,请求体不是合法 JSON 时不会直接抛 400,而是返回None,你再兜底成空字典,接口更健壮。

验证成功的标志有三个:HTTP 状态码 200、Content-Type是application/json、data是数组且每个元素是对象。只要这三点满足,前端和 AI 工具就能正常消费这份数据了。

5. 本篇常见报错排查:500、datetime、401 与连接问题

这一节把你可能撞上的坑集中列一下,都是真实会遇到的。

报错一:TypeError: Object of type 'datetime' is not JSON serializable

这是最经典的。原因就是查询结果里有DATETIME/DATE/TIMESTAMP字段,Python 的datetime对象不是 JSON 原生类型。解决方式就是第 3 节的row_to_dict,逐字段判断并strftime。如果你用json.dumps,也可以传default=str:

json.dumps(data, ensure_ascii=False, default=str)

default=str会让序列化器遇到不认识的对象时调用str(),简单粗暴但有效。不过更推荐显式转换,格式可控。

报错二:接口返回 500,日志里是pymysql.err.OperationalError: (2003, "Can't connect to MySQL server")

数据库连不上。检查三件事:MySQL 服务是否启动(systemctl status mysql或看服务面板);DB_HOST/DB_PORT是否正确,本机就是127.0.0.1:3306;账号密码和库名是否匹配。如果你在容器里跑 Flask、MySQL 在宿主机,127.0.0.1会指向容器自己,要换成宿主机的可达地址。

报错三:pymysql.err.ProgrammingError: (1064, "You have an error in your SQL syntax")

SQL 写错了,常见于表名/字段名拼错,或者用了 MySQL 保留字(比如把字段叫order、key)没加反引号。把拼好的 SQL 打印出来,复制到 MySQL 客户端里跑一遍,能快速定位。

报错四:请求 AI 服务时返回 401 /local proxy failed/OAuth相关错误

这类错误通常不是 Flask 本身的问题,而是调用外部 API 时鉴权或地址配错了。如果你把 Base URL 写成了别的地址,或者 Key 过期、拼写错误,就会 401。统一到 TaoToken 后,Base URL 固定https://taotoken.net/api,Key 从控制台重新生成一个再试。local proxy failed多半是本机网络或地址填错,确认请求地址没有多余路径、没有拼错域名。OAuth 类报错则要检查你用的客户端(比如某些 CLI 工具)是否要求走特定的登录流程,按文档 https://taotoken.net/doc 配置即可。

报错五:RuntimeError: Working outside of application context

一般出现在你把数据库操作抽到独立函数、又在请求上下文外调用时。确保数据库操作都在路由函数内触发,或者用app.app_context()包起来。

报错六:返回的 JSON 中文变成\u5f20\u4eae

jsonify默认不会这样,但如果你手动json.dumps没加ensure_ascii=False,中文就会被转义。加上这个参数即可。

排查通用套路:先看 Flask 控制台 traceback 最后一行,那是真正的错误类型;再看是不是数据库层(pymysql.err)还是序列化层(TypeError);最后用 curl 最小化复现,排除 Postman 配置干扰。

6. 把接口鉴权与 Base URL 统一到 TaoToken 的落地建议

最后说说怎么把这套东西用顺。你的 Flask 接口本身可以加一层简单的鉴权,比如校验请求头里的 token,避免裸奔。同时,把外部 AI 服务的调用也收敛到同一套配置。

一个实用的做法是写个统一的配置模块:

# config.py import os class Config: DB_HOST = os.environ.get("DB_HOST", "127.0.0.1") DB_PORT = int(os.environ.get("DB_PORT", 3306)) DB_USER = os.environ.get("DB_USER", "root") DB_PASSWORD = os.environ.get("DB_PASSWORD", "") DB_NAME = os.environ.get("DB_NAME", "zlnewschema") TAOTOKEN_BASE_URL = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") TAOTOKEN_API_KEY = os.environ.get("TAOTOKEN_API_KEY", "")

Flask 里from config import Config直接用,AI 调用时也读同一份。这样你换环境、换 Key,只动.env。

关于 Key 的获取和管理,统一在控制台操作:https://taotoken.net/console/api-keys 。建议给不同项目建不同的 Key,方便按项目排查用量和吊销。文档在 https://taotoken.net/doc ,模型对话相关的接入看 https://taotoken.net/models ,长期编码或 Agent 场景看 https://taotoken.net/coding-plan 。

如果你用的是 Claude Code 这类工具,它的配置里同样需要 Base URL、Key、Model ID 三件套,把 Base URL 指向https://taotoken.net/api,Key 填控制台生成的,Model ID 按文档填对应模型名即可。这样你的 Flask 数据接口和 AI 工具链共享同一套入口,维护成本一下就降下来了。

实测下来,把地址和 Key 集中管理后,最直观的好处是排错快了:以前 401 要翻五六个文件找是哪配错了,现在只看.env一处。接口返回的 JSON 结构也建议固定成{code, msg, data}三段式,前端和 AI 解析都省心。到这一步,你的 Flask + MySQL 查询接口就算完整跑通了,剩下的就是按业务加字段、加分页、加缓存。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询