python 处理数据库数据 pymysql安装pip install pymysql示例原始写法，cursor.execute() 抛出异常，cursor.close() 和 db_conn.close() 可能不会执行，导致连接泄漏import pymysql import json from pymysql.cursors import DictCursor def connect_sql(base, sql): """ :param base: 要使用的库名 :param sql: sql语句 :return: 返回值 """ db_conn = pymysql.connect( host='localhost', user='root', password='123', database=base, cursorclass=DictCursor, ) # 创建游标 cursor = db_conn.cursor() # 执行sql语句 cursor.execute(sql) # 返回所有数据 db_record = cursor.fetchall() # 关闭游标 cursor.close() # 关闭连接 db_conn.close() # 处理成json return json.dumps(db_record, ensure_ascii=False)使用with优化import pymysql import json from pymysql.cursors import DictCursor # 封装成函数 def connect_sql(base, sql): """ :param base: 要使用的库名 :param sql: sql语句 :return: 返回值 """ # 使用 with 无需手动关闭 with pymysql.connect( host='localhost', user='root', password='123', database=base, # 返回字典 cursorclass=DictCursor, ) as db_conn: with db_conn.cursor() as cursor: # 执行sql语句 cursor.execute(sql) # 返回所有数据 db_record = cursor.fetchall() # 处理成json 方便断言使用 return json.dumps(db_record, ensure_ascii=False)sqlalchemy+pandas 推荐安装pip install sqlalchemy pandas示例import pandas as pd from sqlalchemy import create_engine def connect_sql(base, sql): # 创建SQLAlchemy引擎（格式：mysql + pymysql: // 用户名: 密码 @ 主机 / 数据库名） db_conn = create_engine(f"mysql+pymysql://root:123@localhost/{base}") # 使用 pands 的 pd 执行 sql df = pd.read_sql(sql, db_conn) # json 格式返回值 return df.to_json(orient='records', force_ascii=False)