Python Socket实现C/S架构查询MySQL数据库报错排查

一、SQL语句修正 #

你遇到的变量不存在问题，本质是字符串格式化错误+SQL语法不规范导致的，同时直接拼接SQL存在严重的注入风险，必须改用参数化查询。

错误根源 #

原SQL直接用{ident}和{pays}拼接，既没给字符串类型的paysAmi字段值加单引号，又没处理特殊字符（比如Guinée里的é），导致数据库解析时把Guinée当成了未定义的变量，而非字段值。

修正后的参数化写法 #

不要手动拼接SQL，用MySQL驱动（如mysql-connector-python或pymysql）提供的占位符，自动处理引号和转义：

# 命名占位符（可读性更强）
sql = "SELECT * FROM ami WHERE fk_identreprise = %(ident)s AND paysAmi = %(pays)s;"
# 或位置占位符
sql = "SELECT * FROM ami WHERE fk_identreprise = %s AND paysAmi = %s;"

执行查询时，把参数传给execute()方法，而非拼进SQL：

# 命名参数示例
cursor.execute(sql, {"ident": ident_value, "pays": pays_value})
# 位置参数示例
cursor.execute(sql, (ident_value, pays_value))

这种写法彻底避免语法错误和SQL注入，同时正确解析带特殊字符的pays值。

二、代码优化示例 #

服务端核心逻辑 #

import socket
import threading
import mysql.connector
from mysql.connector import Error
import json

def handle_client(client_socket):
    try:
        # 建议用连接池替代单次连接，提升性能
        conn = mysql.connector.connect(
            host='你的数据库地址',
            database='你的库名',
            user='用户名',
            password='密码'
        )
        if conn.is_connected():
            # 返回字典格式结果，方便后续处理
            cursor = conn.cursor(dictionary=True)

            # 接收客户端JSON格式的参数（比纯字符串更可靠）
            data = client_socket.recv(1024).decode('utf-8')
            params = json.loads(data)
            ident = params.get('ident')
            pays = params.get('pays')

            # 参数化查询
            sql = "SELECT * FROM ami WHERE fk_identreprise = %(ident)s AND paysAmi = %(pays)s;"
            cursor.execute(sql, {"ident": ident, "pays": pays})
            results = cursor.fetchall()

            # 把结果转成JSON发回客户端
            response = json.dumps(results, ensure_ascii=False).encode('utf-8')
            client_socket.send(response)

    except Error as e:
        # 把错误信息返回给客户端，方便排查
        error_msg = json.dumps({"error": str(e)}).encode('utf-8')
        client_socket.send(error_msg)
    finally:
        # 确保资源关闭
        if 'cursor' in locals():
            cursor.close()
        if 'conn' in locals() and conn.is_connected():
            conn.close()
        client_socket.close()

def start_server(host='0.0.0.0', port=9999):
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind((host, port))
    server_socket.listen(5)
    print(f"服务端启动，监听{host}:{port}")

    while True:
        client_sock, addr = server_socket.accept()
        print(f"收到{addr}的连接")
        # 每个客户端请求单独开线程处理
        threading.Thread(target=handle_client, args=(client_sock,)).start()

if __name__ == "__main__":
    start_server()

客户端核心逻辑 #

import socket
import threading
import json

def send_query(client_socket, ident, pays):
    # 用JSON封装参数，避免字符串拼接歧义
    params = {"ident": ident, "pays": pays}
    data = json.dumps(params, ensure_ascii=False).encode('utf-8')
    client_socket.send(data)

def get_result(client_socket):
    response = client_socket.recv(4096).decode('utf-8')
    result = json.loads(response)
    if "error" in result:
        print(f"查询出错: {result['error']}")
    else:
        print("查询结果:")
        for row in result:
            print(row)

def start_client(host='127.0.0.1', port=9999):
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_socket.connect((host, port))

    # 替换为实际要查询的参数
    target_ident = 1001
    target_pays = 'Guinée'

    # 启动发送和接收线程
    send_thread = threading.Thread(target=send_query, args=(client_socket, target_ident, target_pays))
    recv_thread = threading.Thread(target=get_result, args=(client_socket,))

    send_thread.start()
    recv_thread.start()

    send_thread.join()
    recv_thread.join()

    client_socket.close()

if __name__ == "__main__":
    start_client()

三、关键优化点 #

• 参数化查询：解决语法错误和SQL注入问题，自动处理特殊字符。
• JSON传参：避免纯字符串拼接的解析歧义，可靠传递复杂数据。
• 连接池复用：服务端频繁创建/关闭数据库连接会拖慢性能，建议用mysql.connector.pooling实现连接池。
• 线程隔离：每个客户端线程用独立的数据库连接和游标，避免多线程资源冲突。
• 错误反馈：捕获并传递异常信息，方便快速定位问题。

内容的提问来源于stack exchange，提问作者Borris Kouadja NIANGORAN