Python 几种将数据插入到数据库的方法（单行插入、批量插入，SQL Server、MySQL，insert into）
常见的四种插入方式：
一、单行插入（构造insert into代码）
二、批量插入（构造insert into代码）
三、巨量分批次插入（构造insert into代码）
四、pandas.DataFrame插入（支持单行和批量）
示例数据
columnsName = ["SKU", "endCategoryName", "endCategoryID","rootCategoryName", "rootCategoryID", "CategoryTreeName", "CategoryTreeID"
]# 定义值列表
valueList = [[19417978, "Nail Art Tools", 107876, "Health & Beauty", 26395,"Health & Beauty>>Nail Care, Manicure & Pedicure>>Nail Art>>Nail Art Tools", "26395>>47945>>260764>>107876"],[19418353, "Other Fitness, Running & Yoga", 13362, "Sporting Goods", 888,"Sporting Goods>>Fitness, Running & Yoga>>Other Fitness, Running & Yoga", "888>>15273>>13362"],[19418070, "Flags", 43533, "Garden & Patio", 159912, "Garden & Patio>>Décor>>Flags", "159912>>20498>>43533"],[19417996, "Knitting Needles", 71215, "Crafts", 14339,"Crafts>>Needlecrafts & Yarn>>Crocheting & Knitting>>Knitting Needles", "14339>>160706>>3094>>71215"],[19418048, "Binders & Notebooks", 102950, "Home, Furniture & DIY", 11700,"Home, Furniture & DIY>>Stationery & School Equipment>>Binders & Notebooks", "11700>>16092>>102950"]
]零、构造插入引擎的方法
from sqlalchemy import create_engine
# 将特殊字符转成URL编码。若密码中存在特殊字符，则需要先进行URL编码再传入。没有特殊字符可以不用。
from urllib.parse import quote_plus
# pandas.DataFrame用到
import pandas as pd
# 批量插入用到text
from sqlalchemy import textdef myEngine():# driver='mysql' # MySQL的插入驱动driver='mssql' # SQL Server的插入驱动host='10.10.13.11'port=1433user='testUser'password='testUserPassword'database='testDatabase'charset='UTF-8'if driver == 'mysql':conn_str = (f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset='utf8mb4'")elif driver == 'mssql':conn_str = (f"mssql+pymssql://{user}:{password}@{host}:{port}/{database}?charset='UTF-8'")else:raise ValueError("Unsupported driver")engine = create_engine(conn_str, pool_pre_ping=True)return engine
一、单行插入（构造insert into代码）
通过构造insert into代码插入，用 %s 做占位符，在execute方法中将列表数值作为输入参数。
比较灵活，只适合单行插入
def insert_Test_OneByOne():engine = myEngine()tableName = 'test_table'# values = [1478549, "Nail Art Tools", 107876, "Health & Beauty", 26395# , "Health & Beauty>>Nail Care, Manicure & Pedicure>>Nail Art>>Nail Art Tools", "26395>>47945>>260764>>107876"]values = valueList[0]sql = f"""INSERT INTO {tableName} (SKU, endCategoryName, endCategoryID, rootCategoryName, rootCategoryID, CategoryTreeName, CategoryTreeID) VALUES (%s, %s, %s, %s, %s, %s, %s)"""# 执行插入操作try:with engine.connect() as connection:connection.execute(sql, values)print("数据插入成功！")except Exception as e:print(f"插入失败：{str(e)}")二、批量插入（构造insert into代码）
通过构造insert into代码插入，用 :parameter 做占位符，在execute方法中用 text 装饰插入语句，将列表数值作为输入参数。
比较灵活，适合小批量插入。构造相对麻烦。
def insert_Test_ManyByOne():# 小批量插入engine = myEngine()tableName = 'test_table'sql = f"""INSERT INTO {tableName} (SKU, endCategoryName, endCategoryID, rootCategoryName, rootCategoryID, CategoryTreeName, CategoryTreeID) VALUES (:sku, :end_name, :end_id, :root_name, :root_id, :tree_name, :tree_id)"""try:with engine.connect() as connection:connection.execute(text(sql), [{"sku": row[0],"end_name": row[1],"end_id": row[2],"root_name": row[3],"root_id": row[4],"tree_name": row[5],"tree_id": row[6]}for row in valueList # values 应为包含元组的可迭代对象])print("数据插入成功！")except Exception as e:print(f"插入失败：{str(e)}")
三、巨量分批次插入（构造insert into代码）
基本思路是多批次的小批量插入。
比较灵活，适合小批量插入。构造相对麻烦。
def insert_Test_SoManyByOne(longValueList, batchSize=100):# 大量数据的分批插入# 注：占位符和插入的参数名需要一一对应。engine = myEngine()tableName = 'test_table'sql = f"""INSERT INTO {tableName} (SKU, endCategoryName, endCategoryID, rootCategoryName, rootCategoryID, CategoryTreeName, CategoryTreeID) VALUES (:sku, :end_name, :end_id, :root_name, :root_id, :tree_name, :tree_id)"""for i in range(0, len(longValueList), batchSize):batchList = longValueList[i:i + batchSize]try:with engine.connect() as connection:connection.execute(text(sql), [{"sku": row[0],"end_name": row[1],"end_id": row[2],"root_name": row[3],"root_id": row[4],"tree_name": row[5],"tree_id": row[6]}for row in batchList # values 应为包含元组的可迭代对象])print(f"已提交批次 {i // batchSize + 1}/{(len(longValueList) + batchSize-1) // batchSize}")except Exception as e:print(f"插入失败：{str(e)}")
四、pandas.DataFrame插入（支持单行和批量）
基本思路是多批次的小批量插入。
小批量插入，构造容易。
整批插入，报错会整批失败。
def insert_Test_ByDataFrame():# 定义列名dataDF = pd.DataFrame(valueList, columns=columnsName)engine = myEngine()tableName = 'test_table'try:dataDF.to_sql(tableName, con=engine, if_exists='append', index=False)print("数据插入成功！")except Exception as e:print(f"插入失败：{str(e)}")
附录：代码合集
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# Author：Windshield
# Date: 2023/2/6 15:52
# fileName: test.py
from sqlalchemy import create_engine
# 将特殊字符转成URL编码。若密码中存在特殊字符，则需要先进行URL编码再传入。没有可以不需要。
from urllib.parse import quote_plus
import pandas as pd
# 批量插入需要用到text
from sqlalchemy import textdef myEngine():# driver='mysql' # MySQL的插入驱动driver='mssql' # SQL Server的插入驱动host='10.10.13.11'port=1433user='testUser'password='testUserPassword'database='testDatabase'charset='UTF-8'if driver == 'mysql':conn_str = (f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset='utf8mb4'")elif driver == 'mssql':conn_str = (f"mssql+pymssql://{user}:{password}@{host}:{port}/{database}?charset='UTF-8'")else:raise ValueError("Unsupported driver")engine = create_engine(conn_str, pool_pre_ping=True)return enginedef insert_Test_OneByOne():engine = myEngine()tableName = 'test_table'# values = [1478549, "Nail Art Tools", 107876, "Health & Beauty", 26395# , "Health & Beauty>>Nail Care, Manicure & Pedicure>>Nail Art>>Nail Art Tools", "26395>>47945>>260764>>107876"]values = valueList[0]sql = f"""INSERT INTO {tableName} (SKU, endCategoryName, endCategoryID, rootCategoryName, rootCategoryID, CategoryTreeName, CategoryTreeID) VALUES (%s, %s, %s, %s, %s, %s, %s)"""# 执行插入操作try:with engine.connect() as connection:connection.execute(sql, values)print("数据插入成功！")except Exception as e:print(f"插入失败：{str(e)}")def insert_Test_ManyByOne():# 小批量插入engine = myEngine()tableName = 'test_table'sql = f"""INSERT INTO {tableName} (SKU, endCategoryName, endCategoryID, rootCategoryName, rootCategoryID, CategoryTreeName, CategoryTreeID) VALUES (:sku, :end_name, :end_id, :root_name, :root_id, :tree_name, :tree_id)"""try:with engine.connect() as connection:connection.execute(text(sql), [{"sku": row[0],"end_name": row[1],"end_id": row[2],"root_name": row[3],"root_id": row[4],"tree_name": row[5],"tree_id": row[6]}for row in valueList # values 应为包含元组的可迭代对象])print("数据插入成功！")except Exception as e:print(f"插入失败：{str(e)}")def insert_Test_SoManyByOne(longValueList, batchSize=100):# 大量数据的分批插入# 注：占位符和插入的参数名需要一一对应。engine = myEngine()tableName = 'test_table'sql = f"""INSERT INTO {tableName} (SKU, endCategoryName, endCategoryID, rootCategoryName, rootCategoryID, CategoryTreeName, CategoryTreeID) VALUES (:sku, :end_name, :end_id, :root_name, :root_id, :tree_name, :tree_id)"""for i in range(0, len(longValueList), batchSize):batchList = longValueList[i:i + batchSize]try:with engine.connect() as connection:connection.execute(text(sql), [{"sku": row[0],"end_name": row[1],"end_id": row[2],"root_name": row[3],"root_id": row[4],"tree_name": row[5],"tree_id": row[6]}for row in batchList # values 应为包含元组的可迭代对象])print(f"已提交批次 {i // batchSize + 1}/{(len(longValueList) + batchSize-1) // batchSize}")except Exception as e:print(f"插入失败：{str(e)}")def insert_Test_ByDataFrame():# 定义列名dataDF = pd.DataFrame(valueList, columns=columnsName)engine = myEngine()tableName = 'test_table'try:dataDF.to_sql(tableName, con=engine, if_exists='append', index=False)print("数据插入成功！")except Exception as e:print(f"插入失败：{str(e)}")columnsName = ["SKU", "endCategoryName", "endCategoryID","rootCategoryName", "rootCategoryID", "CategoryTreeName", "CategoryTreeID"
]# 定义值列表
valueList = [[19417978, "Nail Art Tools", 107876, "Health & Beauty", 26395,"Health & Beauty>>Nail Care, Manicure & Pedicure>>Nail Art>>Nail Art Tools", "26395>>47945>>260764>>107876"],[19418353, "Other Fitness, Running & Yoga", 13362, "Sporting Goods", 888,"Sporting Goods>>Fitness, Running & Yoga>>Other Fitness, Running & Yoga", "888>>15273>>13362"],[19418070, "Flags", 43533, "Garden & Patio", 159912, "Garden & Patio>>Décor>>Flags", "159912>>20498>>43533"],[19417996, "Knitting Needles", 71215, "Crafts", 14339,"Crafts>>Needlecrafts & Yarn>>Crocheting & Knitting>>Knitting Needles", "14339>>160706>>3094>>71215"],[19418048, "Binders & Notebooks", 102950, "Home, Furniture & DIY", 11700,"Home, Furniture & DIY>>Stationery & School Equipment>>Binders & Notebooks", "11700>>16092>>102950"]
]if __name__ == '__main__':pass
