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)