pymysql 操作示例 - Githubissues


def execute_sql(sql):
    conn = pymysql.connect(host=ip,port=port,user=user,passwd=password,db=db_name,charset = 'utf8',
                           cursorclass = pymysql.cursors.DictCursor)
    cursor = conn.cursor()
    try:
        # 执行sql语句
        cursor.execute(sql)
        #pprint.pprint(sql)
        # 提交执行
        print('Successful')
        conn.commit()

    except Exception as  e:
        # 如果执行sql语句出现问题，则执行回滚操作
        conn.rollback()
        print('Failed')
        print(e)

    finally:
        # 不论try中的代码是否抛出异常，这里都会执行
        # 关闭游标和数据库连接
        cursor.close()
        conn.close()

def insert_update(table_name,data):
    db = pymysql.connect(host=ip,port=port,user=user,passwd=password,db=db_name )
    cursor = db.cursor()
    table = table_name
    keys = ', '.join(data.keys())
    values = ', '.join(['%s'] * len(data))
    sql = 'INSERT INTO {table}({keys}) VALUES ({values}) ON DUPLICATE KEY UPDATE'.format(table=table, keys=keys, values=values)
    update = ','.join([" {key} = %s".format(key=key) for key in data])
    sql += update
    try:
       cursor.execute(sql, tuple(data.values())*2)
       print('Successful')
       db.commit()
    except Exception as  e:
       print('Failed')
       db.rollback()
       print(e)
    cursor.close()
    db.close()

def insert_process(table_name,data):
    db =pymysql.connect(host=ip,port=port,user=user,passwd=password,db=db_name )
    cursor = db.cursor()
    data = data
    table = table_name
    keys = ', '.join(data.keys())
    values = ', '.join(['%s'] * len(data))
    sql = 'INSERT INTO {table}({keys}) VALUES ({values})'.format(table=table, keys=keys, values=values)
    try:
       cursor.execute(sql, tuple(data.values()))
       print('Successful')
       db.commit()
    except:
       print('Failed')
       db.rollback()
    cursor.close()
    db.close()

操作示例

if __name__ == "__main__":
    port = mysql_ss_config.get('port')
    ip = mysql_ss_config.get('ip')
    user = mysql_ss_config.get('user')
    password = mysql_ss_config.get('password')
    db_name ='stock'
    table_name='dim_stock_xueqiu'
    sql_drop_table = """ DROP TABLE  IF EXISTS dim_stock_xueqiu; """;
    execute_sql(sql= sql_drop_table)
    sql_create_table = """
    CREATE TABLE   `dim_stock_xueqiu` (
    `stock_code` VARCHAR (32) NOT NULL  COMMENT 'name',
    `stock_name` VARCHAR (255) CHARACTER SET utf8mb4 DEFAULT NULL ,
    `create_time` datetime DEFAULT CURRENT_TIMESTAMP ,
    `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP , 
     PRIMARY KEY (`stock_code`)
    ) ENGINE = INNODB DEFAULT CHARSET = utf8;
    """.format(table_name=table_name)
    #sql_create_table = sql_create_table.replace('\n','')
    #sql_create_table = sql_create_table.replace('\n', '\r\n')
    sql_create_table = ' '.join([line.strip() for line in sql_create_table.splitlines()]).strip()
    execute_sql(sql= sql_create_table)

ministep / SQL_DataAnalysis

pymysql 操作示例 #29