pymysql进行mysql *** 作

对数据库进行连接并且导出数据库表格

import pandas as pd
import pymysql

# 内部数据库配置, 这是由数仓工程师提供的
sql_config = {
            'host': "x.x.x.x",  # IP地址
            'port': xxxx,  # 端口
            'user': "xxx",  # 用户名
            'password': "xxx",  # 密码
            'database': "xxx",  # 数据库
            # 'charset': 'utf8',
        }


# 打开数据库连接
db = pymysql.connect(**sql_config)

# 使用 cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()


cursor.execute("SELECT * from `t_cy_company`")

# 使用 fetchall() 方法获取全部数据.
data = cursor.fetchall()


# 输出成csv或excel文件，这里是输出csv
# 输出excel使用df.to_excel API即可
cols=cursor.description #类似 desc table_name返回结果
col=[] #创建一个空列表以存放列名
for v in cols:
    col.append(v[0]) #循环提取列名，并添加到col空列表
df = pd.DataFrame(list(data), columns=col)
df.to_excel("./data/xxx.xls")

# 关闭数据库连接
db.close()

​

对数据库进行连接并且写入数据库表格

import pandas as pd
import pymysql


class Action(object):
    """@summary: mysql"""

    def __init__(self):
        mysql = {
            'host': "x.x.x.x",  # IP地址
            'port': xxxx,  # 端口
            'user': "xxx",  # 用户名
            'password': "xxx",  # 密码
            'database': "xxx",  # 数据库
            # 'charset': 'utf8',
        }
        self.db = pymysql.connect(**mysql)
        self.cursor = self.db.cursor()
        print("success_link_mysql")

    def into_database(self, item):
        enterprise_name = str(item[0])
        # food_safe = str(item[1])
        # food_license_details = str(item[2])
        # food_drink = str(item[3])
        # xs_sc_model = str(item[4])
        # Industry_code = str(item[5])
        public_sentiment = str(item[1])
        
        # 同时写入多列
        # sql = '''update t_cytz_company set food_safe="{}", food_license="{}", food_drink="{}", xs_sc_model="{}", Industry_code="{}" where enterprise_name="{}"'''.format(food_safe, food_license_details, food_drink, xs_sc_model, Industry_code, enterprise_name)
        # 写入单列
        sql = '''update t_cytz_company set public_sentiment="{}" where enterprise_name="{}"'''.format(public_sentiment, enterprise_name)
        try:
            self.cursor.execute(sql)
            self.db.commit()
            print(f"{enterprise_name} === 入库成功")
        except Exception as e:
            print(e)
            self.db.rollback()

    def close_mysql(self):
        self.cursor.close()
        self.db.close()
        print("success_close_mysql")


if __name__ == '__main__':
    action = Action()

    data_frame = pd.read_excel('./data/.xls', sheet_name='Sheet1')
    # data = data_frame.loc[:, ['enterprise_name', 'food_safe', 'food_license_details', 'food_drink', 'xs_sc_model', 'Industry_code']].values
    data = data_frame.loc[:, ['id', 'public_sentiment']].values
    for item in data:
        action.into_database(item)

    action.close_mysql()