本文将分享一个基于 Python Flask 框架开发的后端服务实战案例，用于实现对热点新闻或事件数据的管理。核心功能包括数据查询、分页、详情获取以及状态更新，为构建舆情分析或内容管理后台提供了清晰的代码参考。
以下是完整的项目代码 hot_point.py：
from flask import Flask, request, jsonify, json, Blueprint
import pymysql
from pymysql.cursors import DictCursor
app = Flask(__name__)
conf_info = json.load(open('conf.json', 'r'))
DB_CONFIG = {
'host': conf_info["host"],
'port': conf_info["port"],
'user': conf_info["user"],
'password': conf_info["passwd"],
'database': conf_info["db"],
'charset': 'utf8mb4'
}
admin_blueprint = Blueprint('admin_blueprint', __name__, url_prefix='/admin')
@admin_blueprint.route('/hotpoint/<int:id>', methods=['GET'])
def get_hotpoint_detail(id):
try:
connection = pymysql.connect(**DB_CONFIG)
with connection.cursor(DictCursor) as cursor:
sql = "select id, title, content, date_format(release_time,%s) release_time, site, sentiment, visible from hot_point WHERE 1=1 "
params = ['%Y-%m-%d']
if id:
sql += " and id=%s"
params.append(id)
sql += " order by release_time desc"
cursor.execute(sql, params)
hotpoint = cursor.fetchall()
# 返回 JSON 格式数据
return jsonify({
"success": True,
"data": hotpoint,
"message": "success"
})
except Exception as e:
return jsonify({
"success": False,
"error": str(e),
"message": "error"
}), 500
finally:
# 确保连接关闭
if 'connection' in locals() and connection.open:
connection.close()
@admin_blueprint.route('/hotpoint', methods=['PUT'])
def update_hotpoint():
try:
data = request.get_json()
if not data:
return jsonify({
"success": False,
"message": "请传入要更新的数据（JSON 格式）"
}), 400
if not isinstance(data, list):
return jsonify({
"success": False,
"message": "请传入一个 JSON 数组，每个元素包含 id 和 visible"
}), 400
connection = pymysql.connect(**DB_CONFIG)
updated_count = 0
with connection.cursor() as cursor:
for item in data:
record_id = item.get('id')
visible = item.get('visible')
if record_id is None or visible is None:
continue
if not isinstance(visible, int) or visible not in [0, 1]:
continue
sql = "UPDATE hot_point SET visible = %s WHERE id = %s"
cursor.execute(sql, (visible, record_id))
updated_count += cursor.rowcount
connection.commit()
return jsonify({
"success": True,
"updated_count": updated_count,
"message": f"操作成功了 {updated_count} 条记录"
})
except Exception as e:
return jsonify({
"success": False,
"error": str(e),
"message": "操作失败"
}), 500
finally:
if 'connection' in locals() and connection.open:
connection.close()
@admin_blueprint.route('/hotpoint', methods=['POST'])
def hotpoint_list():
conditions = request.get_json()
try:
connection = pymysql.connect(**DB_CONFIG)
if not conditions:
return jsonify({
"success": False,
"message": "请传入查询条件（JSON 格式）"
}), 400
date_from_str = conditions.get('date_from')
date_to_str = conditions.get('date_to')
keyword = conditions.get('keyword')
sentiment_str = conditions.get('sentiment')
visible = conditions.get('visible')
page_num = conditions.get('pageNum', 1) # 第几页，从1开始
page_size = conditions.get('pageSize', 10) # 每页条数
offset = (page_num - 1) * page_size
with connection.cursor(DictCursor) as cursor:
sql = "select id, title, content, date_format(release_time,%s) release_time, site, sentiment, visible from hot_point WHERE 1=1 "
params = ['%Y-%m-%d']
# 日期范围条件
if date_from_str:
sql += " AND release_time >= %s"
params.append(date_from_str)
if date_to_str:
sql += " AND release_time <= %s"
params.append(date_to_str)
if keyword:
sql += " AND (title LIKE %s OR content LIKE %s)"
params.append(f"%{keyword}%")
params.append(f"%{keyword}%")
if sentiment_str:
sql += " AND sentiment = %s"
params.append(sentiment_str)
if visible is not None:
sql += " AND visible = %s"
params.append(visible)
sql += " order by release_time desc"
count_sql = "SELECT COUNT(*) AS total FROM hot_point WHERE 1=1"
count_params = []
if date_from_str:
count_sql += " AND release_time >= %s"
count_params.append(date_from_str)
if date_to_str:
count_sql += " AND release_time <= %s"
count_params.append(date_to_str)
if keyword:
count_sql += " AND (title LIKE %s OR content LIKE %s)"
count_params.append(f"%{keyword}%")
count_params.append(f"%{keyword}%")
if sentiment_str:
count_sql += " AND sentiment = %s"
count_params.append(sentiment_str)
if visible is not None:
count_sql += " AND visible = %s"
count_params.append(visible)
cursor.execute(count_sql, count_params)
total_result = cursor.fetchone()
total = total_result['total'] if total_result else 0
sql += " LIMIT %s OFFSET %s"
params.append(page_size)
params.append(offset)
if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)
hotpoints = cursor.fetchall()
# 返回 JSON 格式数据
return jsonify({
"success": True,
"data": {
"list": hotpoints,
"total": total,
"pageNum": page_num,
"pageSize": page_size
},
"message": "success"
})
except Exception as e:
return jsonify({
"success": False,
"error": str(e),
"message": "error"
}), 500
finally:
# 确保连接关闭
if 'connection' in locals() and connection.open:
connection.close()
@admin_blueprint.route('/interface/yq/data', methods=['POST'])
def yq_data():
orientation = request.args.get("orientation")
try:
connection = pymysql.connect(**DB_CONFIG)
with connection.cursor(DictCursor) as cursor:
sql = "select id, title, content, date_format(release_time,%s) ctime, site, sentiment, visible from hot_point WHERE deleted = 0 and visible=1 "
params = ['%Y-%m-%d']
if orientation:
sentiment_str = '正面' if int(orientation) == 1 else '负面';
sql += " AND sentiment = %s"
params.append(sentiment_str)
sql += " order by release_time desc"
if params:
cursor.execute(sql, params)
else:
cursor.execute(sql)
hotpoint = cursor.fetchall()
# 返回 JSON 格式数据
return jsonify({"result": {
"success": True,
"data": hotpoint,
"message": "success"
}})
except Exception as e:
return jsonify({
"success": False,
"error": str(e),
"message": "error"
}), 500
finally:
# 确保连接关闭
if 'connection' in locals() and connection.open:
connection.close()
if __name__ == '__main__':
app.register_blueprint(admin_blueprint)
app.run(host='0.0.0.0', port=5000, debug=True)
项目配置文件
该服务依赖一个 conf.json 配置文件来读取数据库连接信息。文件内容模板如下，请根据你的实际环境填写：
{
"host": "",
"port": ,
"user": "",
"passwd": "",
"db": "",
"secret_id": "",
"secret_key": "",
"http_host": ""
}
代码详解与设计思路
这段代码实现了一个基于 Flask 的 RESTful API 服务，核心功能是管理存储热点新闻或事件的 hot_point 数据库表。它清晰地展示了配置管理、数据库操作、API 路由设计以及异常处理。
1. 项目初始化与配置
from flask import Flask, request, jsonify, json, Blueprint
import pymysql
from pymysql.cursors import DictCursor
- Flask: 用于创建 Web 应用的核心框架。
- pymysql: Python 连接 MySQL 数据库的驱动。
- DictCursor: 一种数据库游标类型，它使查询结果以字典形式返回（键为列名），便于直接转换为 JSON。
conf_info = json.load(open('conf.json', 'r'))
DB_CONFIG = {
'host': conf_info["host"],
'port': conf_info["port"],
'user': conf_info["user"],
'password': conf_info["passwd"],
'database': conf_info["db"],
'charset': 'utf8mb4'
}
- 从独立的
conf.json 文件加载敏感配置信息，如数据库地址、用户名和密码，这符合将配置与代码分离的最佳实践。
charset 设置为 utf8mb4，确保能够存储包括 Emoji 在内的所有 Unicode 字符。
2. 使用 Blueprint 模块化路由
admin_blueprint = Blueprint('admin_blueprint', __name__, url_prefix='/admin')
- Blueprint 是 Flask 提供的用于组织路由的工具。这里创建了一个名为
admin_blueprint 的蓝图，并设置了 URL 前缀 /admin。这意味着所有在该蓝图下定义的路由都会自动拥有 /admin 的前缀，使得 API 结构更加清晰，也便于未来扩展其他模块。
3. API 接口功能解析
(1) 查询单条热点详情 (GET /admin/hotpoint/<int:id>)
- 功能: 根据传入的路径参数
id，查询特定热点新闻的详细信息。
- SQL 逻辑:
WHERE 1=1 是一个常见的技巧，便于后续动态拼接 AND 条件而无需判断是否是第一个条件。date_format(release_time,%s) release_time 将数据库中的时间字段格式化为 YYYY-MM-DD 的字符串，方便前端显示。- 如果提供了
id，则添加 id=%s 过滤条件。
- 返回: 始终返回一个列表，即使只有一条数据，这保持了接口响应格式的一致性。
(2) 批量更新热点可见状态 (PUT /admin/hotpoint)
- 功能: 批量修改多条热点记录的
visible 字段（通常用于上架/下架操作）。
- 请求体: 期望接收一个 JSON 数组，数组中的每个对象必须包含
id 和 visible (值为 0 或 1)。
- 处理流程:
- 遍历请求数组，对每个有效对象执行
UPDATE 语句。
- 利用
cursor.rowcount 累加实际被修改的行数（注意：如果某条记录原有的 visible 值与新值相同，则不会计数）。
- 返回: 包含成功更新记录数的响应，让调用方明确知道操作结果。
(3) 分页条件查询热点列表 (POST /admin/hotpoint)
- 功能: 这是最复杂的一个接口，支持根据多种条件过滤热点列表，并返回分页结果。
- 支持的查询条件 (通过请求体 JSON 传递):
date_from / date_to: 发布时间范围。keyword: 关键词，同时在标题 (title) 和内容 (content) 中进行模糊匹配。sentiment: 情感倾向（如“正面”、“负面”）。visible: 是否可见状态。pageNum / pageSize: 标准的分页参数。
- 核心设计:
- 动态 SQL 拼接: 根据前端传入的条件，动态地在基础 SQL 语句后添加
WHERE 子句，并使用参数化查询 (params) 来防止 SQL 注入。
- 两次查询: 为了实现高效的分页，首先用相同的条件查询总记录数 (
COUNT(*))，然后再用 LIMIT 和 OFFSET 查询当前页的数据。这是实现分页的经典模式。
- 返回: 一个结构化的响应，包含数据列表 (
list)、总条数 (total)、当前页码和每页大小，方便前端渲染分页组件。
(4) 对外公开数据接口 (POST /admin/interface/yq/data)
- 功能: 提供一个简化的数据接口，通常用于给前端页面或其他外部系统调用，只返回已发布（
visible=1）且未删除（deleted=0）的数据。
- 参数: 通过 URL 查询参数
orientation 过滤情感倾向（1 表示正面，其他值表示负面）。
- 特点: 查询条件固定，不支持复杂过滤和分页，响应格式也可能与内部管理接口略有不同（例如外层多了一个
"result" 键），这体现了内外接口差异化的设计思路。
4. 健壮的错误处理与资源管理
- 异常捕获: 每个接口函数都使用了
try...except...finally 结构。当发生任何未预期的异常时，会捕获并返回一个包含错误信息的 JSON 响应，并附带 HTTP 500 状态码，避免服务崩溃。
- 资源清理: 在
finally 块中，会检查数据库连接是否存在且处于打开状态，然后将其关闭。这是防止数据库连接泄漏的关键，确保即使在处理请求过程中发生异常，连接也能被正确释放。
5. 应用启动入口
if __name__ == '__main__':
app.register_blueprint(admin_blueprint)
app.run(host='0.0.0.0', port=5000, debug=True)
app.register_blueprint(admin_blueprint): 将定义好的蓝图注册到 Flask 应用实例中，使其中定义的所有路由生效。app.run(...): 启动开发服务器。host='0.0.0.0' 表示监听所有公网 IP，debug=True 开启了调试模式（切记：此模式仅用于开发环境，生产环境必须关闭）。
总结与扩展思考
这个案例提供了一个使用 Flask 构建具有基本 CRUD 和分页功能的后端服务的清晰范本。其代码结构清晰，涵盖了 Web 开发中的多个关键点：模块化路由组织、安全的数据库操作、灵活的查询构建、规范的 API 响应以及完善的错误处理。
在实际项目中，你可以在此基础上进一步扩展，例如：
- 增加用户认证与权限控制（如使用 JWT）。
- 对请求参数进行更严格的验证（可使用 Flask-WTF 或 marshmallow）。
- 将数据库操作抽象为模型层（如使用 SQLAlchemy ORM）。
- 添加日志记录。
- 编写单元测试和接口测试。
通过这个实战案例，我们不仅实现了一个功能可用的服务，更展示了一个易于维护和扩展的代码结构。希望这份代码和解析能对你在 云栈社区 中的学习与开发有所帮助。
发表于 10 小时前
|
查看: 0|
回复: 0
