1.引入依赖库
版本:mysql 5.7
pip install mysql-connector
2.连接操作
import mysql.connector
from mysql.connector import errorcode
## 连接数据库
## 第一种连接方式
mysql_db = mysql.connector.connect(
host="192.168.10.222", # 数据库主机地址
port=3306, #端口号
database="test", # 数据库
user="root", # 数据库用户名
passwd="root" # 数据库密码
)
print("是否连接:",mysql_db.is_connected())
print(mysql_db)
## 第二种连接方式
from mysql.connector import (connection)
mysql_connection = connection.MySQLConnection(user='root', password='root',
host='192.168.10.222',
database='test',port=3306)
print("是否连接:",mysql_connection.is_connected())
cursor = mysql_connection.cursor()
## 关闭连接
mysql_connection.close()
## 第三种连接方式
database_config = {
'user': 'root',
'password': 'root',
'host': '192.168.10.222',
'database': 'test',
'raise_on_warnings': True
}
cnx = mysql.connector.connect(**database_config)
print("是否连接:",cnx.is_connected())
cnx.close()
3.创建表操作
import mysql.connector
from mysql.connector import errorcode
## 第二种连接方式
from mysql.connector import connection
## 创建表sql
sql = "create table user_test(user_id VARCHAR(10) not null,user_name VARCHAR(20) null )"
mysql_connection = connection.MySQLConnection(user='root', password='root',
host='192.168.10.222',
database='test',port=3306)
print("是否连接is_connected:",mysql_connection.is_connected())
print("连接信息get_server_info:",mysql_connection.get_server_info())
print("连接信息get_server_version:",mysql_connection.get_server_version())
cursor = mysql_connection.cursor()
try:
## 创建数据库
cursor.execute(sql)
## 提交
mysql_connection.commit()
print("创建成功")
except mysql.connector.Error as error:
## 报错是否存在
if error.errno == errorcode.ER_TABLE_EXISTS_ERROR:
print("already exists.")
else:
print(error.msg)
else:
cursor.close()
## 关闭连接
mysql_connection.close()
4.插入数据操作
import mysql.connector
from mysql.connector import errorcode
## 第二种连接方式
from mysql.connector import connection
## 创建表sql
sql = "INSERT INTO `test`.`user_test` (`user_id`, `user_name`) VALUES ('2', '1')"
mysql_connection = connection.MySQLConnection(user='root', password='root',
host='192.168.10.222',
database='test',port=3306)
print("是否连接is_connected:",mysql_connection.is_connected())
print("连接信息get_server_info:",mysql_connection.get_server_info())
print("连接信息get_server_version:",mysql_connection.get_server_version())
cursor = mysql_connection.cursor()
try:
## 创建数据库
cursor.execute(sql)
## 提交
mysql_connection.commit()
print("创建成功")
except mysql.connector.Error as error:
## 报错是否存在
if error.errno == errorcode.ER_TABLE_EXISTS_ERROR:
print("already exists.")
else:
print(error.msg)
else:
cursor.close()
## 关闭连接
mysql_connection.close()
5.读取数据
import mysql.connector
from mysql.connector import errorcode
## 第二种连接方式
from mysql.connector import connection
from mysql.connector.cursor import MySQLCursor
class MysqlTest:
## 创建表sql
sql = "SELECT * FROM user_test "
## 查询所有数据
def selectAll(self):
mysql_connection = connection.MySQLConnection(user='root', password='root',
host='192.168.10.222',
database='test',port=3306)
print("是否连接is_connected:",mysql_connection.is_connected())
print("连接信息get_server_info:",mysql_connection.get_server_info())
print("连接信息get_server_version:",mysql_connection.get_server_version())
cursor = MySQLCursor(mysql_connection)
try:
## 创建数据库
cursor.execute(self.sql)
## 抓取所有数据
data_rows = cursor.fetchall()
for data in data_rows:
print(data)
print("查询成功")
print("总行数:",cursor.rowcount)
except mysql.connector.Error as error:
## 报错是否存在
if error.errno == errorcode.ER_TABLE_EXISTS_ERROR:
print("already exists.")
else:
print(error.msg)
else:
cursor.close()
# ## 关闭连接
mysql_connection.close()
## 查询一条数据
def selectOne(self):
mysql_connection = connection.MySQLConnection(user='root', password='root',
host='192.168.10.222',
database='test',port=3306)
print("是否连接is_connected:",mysql_connection.is_connected())
print("连接信息get_server_info:",mysql_connection.get_server_info())
print("连接信息get_server_version:",mysql_connection.get_server_version())
cursor = MySQLCursor(mysql_connection)
try:
## 创建数据库
cursor.execute(self.sql)
## 查询一条数据
data_rows = cursor.fetchone()
for data in data_rows:
print(data)
print("查询成功")
print("总行数:",cursor.rowcount)
except mysql.connector.Error as error:
## 报错是否存在
if error.errno == errorcode.ER_TABLE_EXISTS_ERROR:
print("already exists.")
else:
print(error.msg)
else:
# cursor.close()
## 关闭连接
mysql_connection.close()
test = MysqlTest()
print("查一个")
test.selectOne()
test1 = MysqlTest()
print("查所有")
test1.selectAll()
以上为python 连接数据库的基本操作 主要使用 execute 方法