连接MySQL数据库并插入数据是Python中常见的任务，通常可以通过mysql-connector-python库来实现。下面是一个简单的示例脚本，展示了如何连接MySQL数据库并插入数据：

1.安装mysql-connector-python库

如果你还没有安装这个库，可以通过pip安装：

1

pip install mysql-connector-python

2.编写Python脚本

下面是一个示例脚本，它连接到MySQL数据库，创建一个新表（如果不存在），然后插入一些数据。

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

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")

3.运行脚本

将上述脚本保存为一个.py文件，然后运行它。确保你已经正确设置了数据库的连接参数（如主机、用户名、密码和数据库名）。

这个脚本首先尝试连接到MySQL数据库，如果连接成功，它会检查是否存在一个名为employees的表，如果不存在则创建这个表。然后，脚本会向这个表中插入一行数据。最后，它会关闭数据库连接。

4.方法补充

python实现连接mysql脚本

完整代码:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

#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]

python脚本连接mysql数据库的方法

下面介绍了如何利用Python的pymysql库进行数据库登录、执行SQL查询并获取所有数据的过程，包括连接数据库、执行SQL语句、获取查询结果及关闭连接等关键步骤

完整代码：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

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() ## 关闭连接

python连接mysql工具类+进行数据驱动+配置模块

即：将python连接mysql相关的代码封装起来，形成一个工具类，在需要使用的时候调用

准备工作：1、key_demo——2、cases|tool——3、cases/case.py|tool/MyDB.py

1、MyDB.py工具类

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

# -*- 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()

2、case.py数据驱动

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

# -*- 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()

这样写的插入语句会被调用多次，要注意@data中执行的操作具体执行次数

换了一个写法，先插入，再查询

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

# -*- 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()

代码编写的时候有几个注意事项：

1、只执行一条语句，返回结果也只有一个，不需要@unpack解析

2、MyDB.py中定义有返回值，还有ddt中定义的语法return func(self, *args, **kwargs),不能缺少参数请确保在运行脚本之前，你的MySQL服务器正在运行，并且你已经正确配置了数据库的访问权限。

# 数据库 # python # mysql
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

深入理解 AI Agent · AGENT #02：Agent 的四大核心机制，拆开来看每个都在干什么

avatar AI Agent技术社区

MCP协议开发实战：从零搭建AI Agent工具链

本文介绍了MCP协议在AI工具链中的应用与实践。MCP协议旨在解决异构工具集成难题，为AI模型提供标准化、安全的工具调用接口。文章详细解析了MCP的核心架构，包括Server、Client、Transport等组件，以及通信流程与安全模型。通过实战示例，指导读者开发计算器和文件系统工具Server，并集成到Claude等AI客户端中。进阶部分展示了如何构建复杂工具链，包括API集成、数据库查询和多

avatar AI Agent技术社区
cover

AI Tool 设计：粒度、参数与错误恢复怎么做

avatar AI Agent技术社区