创建数据的完整代码如下：
from contextlib import contextmanager
import mysql.connector
from mysql.connector import Error
class DatabaseConfig:
"""数据库配置类，集中管理连接参数"""
HOST = '192.168.116.70' # 数据库服务器地址
PORT = 3306 # 数据库端口号
DATABASE = 'python_mysql_demo' # 数据库名
USER = 'root' # 数据库用户名
PASSWORD = 'aaaaaa' # 数据库密码
CHARSET = 'utf8mb4' # 字符集，该字符集可支持emoji表情
class Database:
"""数据库操作封装"""
def __init__(self, config=DatabaseConfig):
self.config = config
self.connection = None
def connect(self):
"""建立连接"""
self.connection = mysql.connector.connect(
host=self.config.HOST,
port=self.config.PORT,
database=self.config.DATABASE,
user=self.config.USER,
password=self.config.PASSWORD,
charset=self.config.CHARSET,
autocommit=False
)
return self.connection
def close(self):
"""关闭连接"""
if self.connection and self.connection.is_connected():
self.connection.close()
def __enter__(self):
"""
进入上下文管理器，建立数据库连接
在 with 语句入口处被调用，建立连接并返回实例本身
"""
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""
退出上下文管理器，关闭数据库连接
"""
self.close()
def insert_student(db, name, age, email):
"""插入单条记录，使用参数化替换防止 SQL 注入"""
sql = """
INSERT INTO students (name, age, email)
VALUES (%s, %s, %s)
"""
cursor = db.connection.cursor()
cursor.execute(sql, (name, age, email))
db.connection.commit()
return cursor.lastrowid # 返回自增 ID
def batch_insert_students(db, records):
"""批量插入，使用 executemany 提升性能"""
sql = "INSERT INTO students (name, age, email) VALUES (%s, %s, %s)"
cursor = db.connection.cursor()
cursor.executemany(sql, records) # 批量执行
db.connection.commit()
return cursor.rowcount # 受影响行数
# 使用示例
with Database() as db:
# 单条插入
new_id = insert_student(db, '张三', 21, 'zhangsan@example.com')
print("新记录 ID:", new_id)
# 批量插入
data = [
('李四', 22, 'lisi@example.com'),
('王五', 23, 'wangwu@example.com'),
('赵六', 20, 'zhaoliu@example.com'),
]
count = batch_insert_students(db, data)
print("批量插入了", count, "条记录")
川公网安备51010802032098