当前位置：首页 > news >正文

板凳-------Mysql cookbook学习（十二--------3_1)

news 2025/7/22 10:51:17

顺序表：202507212 16：40 p74
存储表信息的单元与元素存储区分离式结构

import mysql.connector
from mysql.connector import Errorclass SeparatedList:def __init__(self, initial_capacity=100000):self.metadata = {'capacity': initial_capacity,'length': 0,'storage': [None] * initial_capacity}def insert(self, e, pos):if pos < 0 or pos > self.metadata['length']:raise IndexError("插入位置越界")if self.metadata['length'] == self.metadata['capacity']:self._expand_storage()for i in range(self.metadata['length'], pos, -1):self.metadata['storage'][i] = self.metadata['storage'][i-1]self.metadata['storage'][pos] = eself.metadata['length'] += 1def delete(self, pos):if pos < 0 or pos >= self.metadata['length']:raise IndexError("删除位置越界")for i in range(pos, self.metadata['length'] - 1):self.metadata['storage'][i] = self.metadata['storage'][i+1]self.metadata['storage'][self.metadata['length'] - 1] = Noneself.metadata['length'] -= 1def retrieve(self, pos):if pos < 0 or pos >= self.metadata['length']:raise IndexError("访问位置越界")return self.metadata['storage'][pos]def length(self):return self.metadata['length']def is_empty(self):return self.metadata['length'] == 0def print_list(self, limit=10):print(f"表信息: 容量={self.metadata['capacity']}, 长度={self.metadata['length']}")if self.metadata['length'] <= limit:print("元素存储区:", [elem for elem in self.metadata['storage'] if elem is not None])else:print(f"前{limit}个元素:", [self.metadata['storage'][i] for i in range(limit)])print(f"... 省略 {self.metadata['length'] - limit} 个元素 ...")def _expand_storage(self):new_capacity = self.metadata['capacity'] * 2new_storage = [None] * new_capacityfor i in range(self.metadata['length']):new_storage[i] = self.metadata['storage'][i]self.metadata['storage'] = new_storageself.metadata['capacity'] = new_capacityprint(f"存储区已扩容至 {new_capacity}")def get_employees_data(host, database, user, password, limit=500000):connection = Noneemployees_data = []try:connection = mysql.connector.connect(host=host,database=database,user=user,password=password)if connection.is_connected():cursor = connection.cursor(dictionary=True)query = f"""SELECT emp_no, first_name, last_name, gender, hire_date FROM employees LIMIT {limit}"""cursor.execute(query)records = cursor.fetchall()print(f"成功从数据库获取 {len(records)} 条员工数据")for record in records:emp_info = (record['emp_no'],f"{record['first_name']} {record['last_name']}",record['gender'],record['hire_date'].strftime('%Y-%m-%d'))employees_data.append(emp_info)except Error as e:print(f"数据库操作错误: {e}")finally:if connection and connection.is_connected():cursor.close()connection.close()print("数据库连接已关闭")return employees_dataif __name__ == "__main__":# 数据库配置（确保参数正确且加引号）DB_CONFIG = {'host': 'localhost','database': 'employees','user': 'root','password': 'root'  # 替换为你的MySQL密码}slist = SeparatedList(initial_capacity=100000)print("初始列表:")slist.print_list()# 从数据库获取数据（以下代码缩进统一为4个空格）print("\n从数据库获取员工数据...")employees = get_employees_data(host=DB_CONFIG['host'],database=DB_CONFIG['database'],user=DB_CONFIG['user'],password=DB_CONFIG['password'],limit=500000)# 将数据插入列表（缩进统一）print("\n将员工数据插入到列表中...")for i, emp in enumerate(employees):slist.insert(emp, i)if (i + 1) % 100000 == 0:print(f"已插入 {i + 1} 条数据，当前列表长度: {slist.length()}")# 测试检索# 测试检索功能print("\n测试检索功能:")total = slist.length()  # 获取实际数据总量if total > 0:# 第100条（索引99）if 99 < total:print("第100条数据:", slist.retrieve(99))else:print("数据不足100条")# 第100000条（索引99999）if 99999 < total:print("第100000条数据:", slist.retrieve(99999))else:print("数据不足100000条")# 第300000条（索引299999，根据实际总量调整）if 299999 < total:print("第300000条数据:", slist.retrieve(299999))else:print(f"数据不足300000条，当前总量为{total}条")# 测试删除print("\n测试删除功能:")if slist.length() > 1000:slist.delete(1000)print("删除第1001条数据后，该位置的数据变为:", slist.retrieve(1000))# 测试插入print("\n测试插入功能:")new_employee = (999999, "John Doe", "M", "2023-01-01")slist.insert(new_employee, 0)print("在开头插入新员工后，第1条数据为:", slist.retrieve(0))# 最终状态print("\n最终列表状态:")slist.print_list()print("列表长度:", slist.length())print("列表是否为空:", slist.is_empty())
====================================================
初始列表:
表信息: 容量=100000, 长度=0
元素存储区: []从数据库获取员工数据...
成功从数据库获取 300024 条员工数据
数据库连接已关闭将员工数据插入到列表中...
已插入 100000 条数据，当前列表长度: 100000
存储区已扩容至 200000
已插入 200000 条数据，当前列表长度: 200000
存储区已扩容至 400000
已插入 300000 条数据，当前列表长度: 300000测试检索功能:
第100条数据: (10100, 'Hironobu Haraldson', 'F', '1987-09-21')
第100000条数据: (110000, 'Supot Herath', 'M', '1987-07-07')
第300000条数据: (499975, 'Masali Chorvat', 'M', '1992-01-23')测试删除功能:
删除第1001条数据后，该位置的数据变为: (11002, 'Bluma Ulupinar', 'M', '1996-12-23')测试插入功能:
在开头插入新员工后，第1条数据为: (999999, 'John Doe', 'M', '2023-01-01')最终列表状态:
表信息: 容量=400000, 长度=300024
前10个元素: [(999999, 'John Doe', 'M', '2023-01-01'), (10001, 'Georgi Facello', 'M', '1986-06-26'), (10002, 'Bezalel Simmel', 'F', '1985-11-21'), (10003, 'Parto Bamford', 'M', '1986-08-28'), (10004, 'Chirstian Koblick', 'M', '1986-12-01'), (10005, 'Kyoichi Maliniak', 'M', '1989-09-12'), (10006, 'Anneke Preusig', 'F', '1989-06-02'), (10007, 'Tzvetan Zielinski', 'F', '1989-02-10'), (10008, 'Saniya Kalloufi', 'M', '1994-09-15'), (10009, 'Sumant Peac', 'F', '1985-02-18')]
... 省略 300014 个元素 ...
列表长度: 300024
列表是否为空: False

import mysql.connector
from mysql.connector import Error
import configparserclass SeparatedList:# （类定义部分保持不变）def __init__(self, initial_capacity=100000):self.metadata = {'capacity': initial_capacity,'length': 0,'storage': [None] * initial_capacity}def insert(self, e, pos):if pos < 0 or pos > self.metadata['length']:raise IndexError("插入位置越界")if self.metadata['length'] == self.metadata['capacity']:self._expand_storage()for i in range(self.metadata['length'], pos, -1):self.metadata['storage'][i] = self.metadata['storage'][i-1]self.metadata['storage'][pos] = eself.metadata['length'] += 1def delete(self, pos):if pos < 0 or pos >= self.metadata['length']:raise IndexError("删除位置越界")for i in range(pos, self.metadata['length'] - 1):self.metadata['storage'][i] = self.metadata['storage'][i+1]self.metadata['storage'][self.metadata['length'] - 1] = Noneself.metadata['length'] -= 1def retrieve(self, pos):if pos < 0 or pos >= self.metadata['length']:raise IndexError("访问位置越界")return self.metadata['storage'][pos]def length(self):return self.metadata['length']def is_empty(self):return self.metadata['length'] == 0def print_list(self, limit=10):print(f"表信息: 容量={self.metadata['capacity']}, 长度={self.metadata['length']}")if self.metadata['length'] <= limit:print("元素存储区:", [elem for elem in self.metadata['storage'] if elem is not None])else:print(f"前{limit}个元素:", [self.metadata['storage'][i] for i in range(limit)])print(f"... 省略 {self.metadata['length'] - limit} 个元素 ...")def _expand_storage(self):new_capacity = self.metadata['capacity'] * 2new_storage = [None] * new_capacityfor i in range(self.metadata['length']):new_storage[i] = self.metadata['storage'][i]self.metadata['storage'] = new_storageself.metadata['capacity'] = new_capacityprint(f"存储区已扩容至 {new_capacity}")

可以使用 Python 的configparser模块来读取config.ini文件并获取数据库配置信息，以替换原有的字典配置。具体操作步骤如下：
1. 创建 config.ini 文件
在项目根目录或合适的位置创建一个名为config.ini的文件，内容如下：
ini
[Database]
host = localhost
database = employees
user = root
password = root  ; 替换为你的MySQL密码[Database]是配置节（section）名称，可以根据喜好命名，括号内的名称需与后续代码中读取时使用的名称一致。
def get_employees_data(host, database, user, password, limit=500000):# （函数代码保持不变）connection = Noneemployees_data = []try:connection = mysql.connector.connect(host=host,database=database,user=user,password=password)if connection.is_connected():cursor = connection.cursor(dictionary=True)query = f"""SELECT emp_no, first_name, last_name, gender, hire_date FROM employees LIMIT {limit}"""cursor.execute(query)records = cursor.fetchall()print(f"成功从数据库获取 {len(records)} 条员工数据")for record in records:emp_info = (record['emp_no'],f"{record['first_name']} {record['last_name']}",record['gender'],record['hire_date'].strftime('%Y-%m-%d'))employees_data.append(emp_info)except Error as e:print(f"数据库操作错误: {e}")finally:if connection and connection.is_connected():cursor.close()connection.close()print("数据库连接已关闭")return employees_dataif __name__ == "__main__":# 修正1：路径用原始字符串（加 r 前缀）config = configparser.ConfigParser()config.read(r'D:\sql\Mysql_learning\config.ini')  # 关键：解决路径转义问题# 修正2：统一缩进（4个空格，与外层代码对齐）DB_CONFIG = {'host': config.get('Database', 'host'),'database': config.get('Database', 'database'),'user': config.get('Database', 'user'),'password': config.get('Database', 'password')}slist = SeparatedList(initial_capacity=100000)print("初始列表:")slist.print_list()print("\n从数据库获取员工数据...")employees = get_employees_data(host=DB_CONFIG['host'],database=DB_CONFIG['database'],user=DB_CONFIG['user'],password=DB_CONFIG['password'],limit=500000)# 将数据插入列表（缩进统一为4个空格）print("\n将员工数据插入到列表中...")for i, emp in enumerate(employees):slist.insert(emp, i)if (i + 1) % 100000 == 0:print(f"已插入 {i + 1} 条数据，当前列表长度: {slist.length()}")# 测试检索（缩进统一）print("\n测试检索功能:")total = slist.length()if total > 0:if 99 < total:print("第100条数据:", slist.retrieve(99))else:print("数据不足100条")if 99999 < total:print("第100000条数据:", slist.retrieve(99999))else:print("数据不足100000条")if 299999 < total:print("第300000条数据:", slist.retrieve(299999))else:print(f"数据不足300000条，当前总量为{total}条")# 测试删除print("\n测试删除功能:")if slist.length() > 1000:slist.delete(1000)print("删除第1001条数据后，该位置的数据变为:", slist.retrieve(1000))# 测试插入print("\n测试插入功能:")new_employee = (999999, "John Doe", "M", "2023-01-01")slist.insert(new_employee, 0)print("在开头插入新员工后，第1条数据为:", slist.retrieve(0))# 最终状态print("\n最终列表状态:")slist.print_list()print("列表长度:", slist.length())print("列表是否为空:", slist.is_empty())初始列表:
表信息: 容量=100000, 长度=0
元素存储区: []从数据库获取员工数据...
数据库操作错误: 1045 (28000): Access denied for user 'root'@'localhost' (using password: YES)将员工数据插入到列表中...测试检索功能:测试删除功能:测试插入功能:
在开头插入新员工后，第1条数据为: (999999, 'John Doe', 'M', '2023-01-01')最终列表状态:
表信息: 容量=100000, 长度=1
元素存储区: [(999999, 'John Doe', 'M', '2023-01-01')]
列表长度: 1
列表是否为空: False