系统概述
该脚本主要实现以下功能:
- 自动连接到 MySQL 数据库
- 创建测试数据库和表结构
- 批量插入 100 条初始测试数据
- 每分钟自动插入一条新数据
- 提供数据统计和最新记录查询功能
完整代码实现
#!/usr/bin/env python3
import pymysql
import time
import random
from datetime import datetime
import sys
import threading
# 数据库连接配置
DB_CONFIG = {
'host': 'localhost',
'user': 'root',
'password': 'your_password',
'database': 'test_db'
}
def connect_db():
"""建立数据库连接"""
try:
conn = pymysql.connect(**DB_CONFIG)
return conn
except Exception as e:
print(f"连接失败:{e}")
return None
def create_schema(conn):
"""创建数据库和表结构"""
cursor = conn.cursor()
# 创建数据库(如果不存在)
cursor.execute("CREATE DATABASE IF NOT EXISTS test_db")
cursor.execute("USE test_db")
# 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS data_log (
id INT AUTO_INCREMENT PRIMARY KEY,
content VARCHAR(255),
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
cursor.close()
():
cursor = conn.cursor()
sql =
values = [() i (count)]
cursor.executemany(sql, values)
conn.commit()
()
cursor.close()
():
cursor = conn.cursor()
sql =
value =
cursor.execute(sql, (value,))
conn.commit()
()
cursor.close()
():
cursor = conn.cursor()
cursor.execute()
result = cursor.fetchone()
()
cursor.close()
():
conn = connect_db()
conn:
sys.exit()
create_schema(conn)
batch_insert(conn, )
():
:
insert_single(conn)
time.sleep()
thread = threading.Thread(target=schedule_task, daemon=)
thread.start()
()
:
:
time.sleep()
query_latest(conn)
KeyboardInterrupt:
()
conn.close()
__name__ == :
main()

