Python 与 MySQL 数据库交互实践：自动化数据插入系统

系统概述

该脚本主要实现以下功能：

1. 自动连接到 MySQL 数据库
2. 创建测试数据库和表结构
3. 批量插入 100 条初始测试数据
4. 每分钟自动插入一条新数据
5. 提供数据统计和最新记录查询功能

完整代码实现

#!/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()