使用Python Connector按通用词查询数据——避免代码重复

合并SystemVar表的多次查询 #

你原来的三次查询都是从同一张表取不同name的val，完全可以用IN子句一次搞定，减少数据库交互次数：

cur = connection.cursor()
# 一次性查询所有需要的系统变量
cur.execute("""
    SELECT name, val 
    FROM jDev.SystemVar 
    WHERE name IN (%s, %s, %s)
""", ("pr_screen_height", "pr_screen_width", "pr_screen_distance"))

# 把结果转成字典，方便直接按name取数值
system_vars = {row[0]: int(row[1]) for row in cur.fetchall()}

# 直接从字典里拿变量，不用再写重复的查询代码
screen_height_mm = system_vars["pr_screen_height"]
screen_width_mm = system_vars["pr_screen_width"]
screen_distance_mm = system_vars["pr_screen_distance"]

如果你的变量都是以pr_screen_开头的，想用LIKE批量匹配（后续加同前缀变量不用改SQL），可以这么写：

cur.execute("""
    SELECT name, val 
    FROM jDev.SystemVar 
    WHERE name LIKE 'pr_screen_%'
""")
system_vars = {row[0]: int(row[1]) for row in cur.fetchall()}
# 同样从字典里取对应的变量就行

简化BMsg表的查询 #

同理，BMsg的两次查询也可以合并成一次，再按type分组：

cur.execute("""
    SELECT type, msg 
    FROM jDev.BMsg 
    WHERE type IN (%s, %s)
""", ("DeviceMessage", "ZeroMessage"))

# 按消息类型分组存储
msg_groups = {}
for msg_type, msg_content in cur.fetchall():
    if msg_type not in msg_groups:
        msg_groups[msg_type] = []
    msg_groups[msg_type].append(msg_content)

dev_msgs = msg_groups["DeviceMessage"]
zero_msgs = msg_groups["ZeroMessage"]

封装通用函数彻底减少重复 #

要是以后还要写类似的查询，不如把重复的逻辑封装成函数，调用起来更省事：

def get_key_value_pairs(cur, table, key_col, value_col, key_list):
    """批量查询键值对数据，返回字典"""
    # 生成对应数量的占位符
    placeholders = ", ".join(["%s"] * len(key_list))
    query = f"SELECT {key_col}, {value_col} FROM {table} WHERE {key_col} IN ({placeholders})"
    cur.execute(query, key_list)
    return {row[0]: row[1] for row in cur.fetchall()}

# 查询系统变量
system_vars = get_key_value_pairs(cur, "jDev.SystemVar", "name", "val", 
                                  ["pr_screen_height", "pr_screen_width", "pr_screen_distance"])
screen_height_mm = int(system_vars["pr_screen_height"])

# 再封装一个按列分组返回列表的函数
def get_grouped_lists(cur, table, group_col, value_col, group_list):
    placeholders = ", ".join(["%s"] * len(group_list))
    query = f"SELECT {group_col}, {value_col} FROM {table} WHERE {group_col} IN ({placeholders})"
    cur.execute(query, group_list)
    groups = {}
    for group_val, item in cur.fetchall():
        groups.setdefault(group_val, []).append(item)
    return groups

# 查询消息
msg_groups = get_grouped_lists(cur, "jDev.BMsg", "type", "msg", ["DeviceMessage", "ZeroMessage"])
dev_msgs = msg_groups["DeviceMessage"]
zero_msgs = msg_groups["ZeroMessage"]

这样不管以后加多少类似的查询，只要调用函数传参数就行，不用重复写执行SQL、处理结果的代码。

内容的提问来源于stack exchange，提问作者ju09