Python脚本实现mysql数据库连接并插入数据

pip install mysql-connector-python

import mysql.connector
from mysql.connector import Error

try:
    # 连接到MySQL数据库
    connection = mysql.connector.connect(
        host='localhost',       # 数据库服务器地址
        user='your_username',   # 数据库用户名
        password='your_password', # 数据库密码
        database='your_database_name'  # 数据库名
    )

    if connection.is_connected():
        print("Connected to MySQL database")

        # 创建一个cursor对象
        cursor = connection.cursor()

        # 创建表
        create_table_query = """
        CREATE TABLE IF NOT EXISTS employees (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(255) NOT NULL,
            salary INT,
            department VARCHAR(255)
        );
        """
        cursor.execute(create_table_query)
        print("Table created or already exists")

        # 插入数据
        insert_query = """
        INSERT INTO employees (name, salary, department)
        VALUES (%s, %s, %s);
        """
        data = ("John Doe", 70000, "Finance")
        cursor.execute(insert_query, data)
        connection.commit()
        print("Data inserted successfully")

except Error as e:
    print("Error while connecting to MySQL", e)

finally:
    # 关闭cursor和连接
    if connection.is_connected():
        cursor.close()
        connection.close()
        print("MySQL connection is closed")

#encoding=gbk
import  sys
import  MySQLdb
 
 
conn=MySQLdb.Connection('127.0.0.1','root','123456','job',charset='gbk')
cur=conn.cursor()
cur.execute("select * from demo_jobs_store")
conn.commit()
conn.close()
 
cur.scroll(0)
row1=cur.fetchone()
print row1[3]

import pymysql
 
## 登录账号
host = "host_name"
user = "user_name"
password = "password"
database = "database_name"
port = port
 
conn = pymysql.connect(host, user, passwd, db, port) ## 登录数据库，返回连接
 
cursor = conn.cursor()
sql = "select var1, ..., varn from table_name where ..."
cursor.execute(sql)
 
data = cursor.fetchall() ##data为以元组为元素的元组
conn.commit()
cursor.close()
conn.close() ## 关闭连接

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/18 15:20
# @Function:
import pymysql
 
 
class my_db:
    '''
    动作类：获取数据连接，连接ip,端口，账号密码。。
    '''
 
    # 构造函数
    def __init__(self):
        try:
            self.dbconn = pymysql.connect(host='127.0.0.1', port=3306, user='root', password='root',
                                          database='test_cases', charset='utf8')
        except Exception as e:
            print('初始化数据库连接失败：%s' % e)
 
    def close(self):
        self.dbconn.close()
 
    # query=查询语句
    def select_record(self, query):
        # 查询数据
        print('query:%s' % query)
        try:
            # 建立游标
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            result = db_cursor.fetchall()
            return result
        except Exception as e:
            print('数据库查询数据失败：%s' % e)
            db_cursor.close()
            exit()
 
    # 插入
    def execute_insert(self, query):
        print('query:%s' % query)
        try:
            # 建立游标
            db_cursor = self.dbconn.cursor()
            db_cursor.execute(query)
            db_cursor.execute('commit')
            return True
        except Exception as e:
            print('数据库插入数据失败：%s' % e)
            # 事务回滚
            db_cursor.execute('rollback')
            db_cursor.close()
            exit()

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/7 15:07
# @Function:
# 针对数据库中的数据做数据驱动
import time
import unittest
from ddt import ddt, data, unpack
 
from key_demo.tool.MyDB import my_db
 
# 必须要实例化，不然会报TypeError: select_record() missing 1 required positional argument: 'query'
testdb = my_db()
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
 
 
@ddt
class case(unittest.TestCase):
    # @data(*my_db().select_record('select weaid,success from weather'))
    @data(*testdb.select_record('select weaid,success from weather'))
    @unpack
    def test_1(self, weaid, success):
        print(weaid, success)
        testdb.execute_insert('insert into weather(weaid,success,cre_time) values ("2","2","%s")' % now)
 
 
if __name__ == '__main__':
    unittest.main()

# -*- coding: utf-8 -*-
# @Author  : hxy
# @Time    : 2022/1/7 15:07
# @Function:
# 针对数据库中的数据做数据驱动
import time
import unittest
from ddt import ddt, data, unpack
 
from key_demo.tool.MyDB import my_db
 
# 必须要实例化，不然会报TypeError: select_record() missing 1 required positional argument: 'query'
testdb = my_db()
now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
 
 
@ddt
class case(unittest.TestCase):
    # @data(*my_db().select_record('select weaid,success from weather'))
    @data(testdb.execute_insert('insert into weather(weaid,success,cre_time)values("2","2","%s")' % now))
    # @unpack
    def test_1(self,t):
        print(t)
        res = testdb.select_record('select weaid,success from weather')
        print(res)
 
 
if __name__ == '__main__':
    unittest.main()