Python脚本实现mysql数据库连接并插入数据
<div id="navCategory"><h5 class="catalogue">目录</h5><ul class="first_class_ul"><li><a href="#_label0">1.安装mysql-connector-python库</a></li><li><a href="#_label1">2.编写Python脚本</a></li><li><a href="#_label2">3.运行脚本</a></li><li><a href="#_label3">4.方法补充</a></li><ul class="second_class_ul"><li><a href="#_lab2_3_0">python实现连接mysql脚本</a></li><li><a href="#_lab2_3_1">python脚本连接mysql数据库的方法</a></li><li><a href="#_lab2_3_2">python连接mysql工具类+进行数据驱动+配置模块</a></li></ul></ul></div><p>连接MySQL数据库并插入数据是Python中常见的任务，通常可以通过<code>mysql-connector-python</code>库来实现。下面是一个简单的示例脚本，展示了如何连接MySQL数据库并插入数据：</p><p class="maodian"><a name="_label0"></a></p><h2>1.安装mysql-connector-python库</h2>
<p>如果你还没有安装这个库，可以通过pip安装：</p>
<div class="jb51code"><pre class="brush:bash;">pip install mysql-connector-python
</pre></div>
<p class="maodian"><a name="_label1"></a></p><h2>2.编写Python脚本</h2>
<p>下面是一个示例脚本，它连接到MySQL数据库，创建一个新表（如果不存在），然后插入一些数据。</p>
<div class="jb51code"><pre class="brush:py;">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")
</pre></div>
<p class="maodian"><a name="_label2"></a></p><h2>3.运行脚本</h2>
<p>将上述脚本保存为一个<code>.py</code>文件，然后运行它。确保你已经正确设置了数据库的连接参数（如主机、用户名、密码和数据库名）。</p>
<p>这个脚本首先尝试连接到MySQL数据库，如果连接成功，它会检查是否存在一个名为<code>employees</code>的表，如果不存在则创建这个表。然后，脚本会向这个表中插入一行数据。最后，它会关闭数据库连接。</p>
<p class="maodian"><a name="_label3"></a></p><h2>4.方法补充</h2>
<p class="maodian"><a name="_lab2_3_0"></a></p><h3>python实现连接mysql脚本</h3>
<p>完整代码:</p>
<div class="jb51code"><pre class="brush:py;">#encoding=gbk
importsys
importMySQLdb
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</pre></div>
<p class="maodian"><a name="_lab2_3_1"></a></p><h3>python脚本连接mysql数据库的方法</h3>
<p>下面介绍了如何利用Python的pymysql库进行数据库登录、执行SQL查询并获取所有数据的过程，包括连接数据库、执行SQL语句、获取查询结果及关闭连接等关键步骤</p>
<p>完整代码：</p>
<div class="jb51code"><pre class="brush:py;">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() ## 关闭连接</pre></div>
<p class="maodian"><a name="_lab2_3_2"></a></p><h3>python连接mysql工具类+进行数据驱动+配置模块</h3>
<p>即：将python连接mysql相关的代码封装起来，形成一个工具类，在需要使用的时候调用</p>
<p>准备工作：1、key_demo——2、cases|tool——3、cases/case.py|tool/MyDB.py</p>
<p><strong>1、MyDB.py工具类</strong></p>
<div class="jb51code"><pre class="brush:py;"># -*- 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()</pre></div>
<p><strong>2、case.py数据驱动</strong></p>
<div class="jb51code"><pre class="brush:py;"># -*- 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()</pre></div>
<p>这样写的插入语句会被调用多次，要注意@data中执行的操作具体执行次数</p>
<p><strong>换了一个写法，先插入，再查询</strong></p>
<div class="jb51code"><pre class="brush:py;"># -*- 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()</pre></div>
<p>代码编写的时候有几个注意事项：</p>
<p>1、只执行一条语句，返回结果也只有一个，不需要@unpack解析</p>
<p>2、MyDB.py中定义有返回值，还有ddt中定义的语法return func(self, *args, **kwargs),不能缺少参数</p>
<p>请确保在运行脚本之前，你的MySQL服务器正在运行，并且你已经正确配置了数据库的访问权限。</p>
頁:
[1]