import pymysql
# 链接数据库
conn = pymysql.connect(
host='10.5.6.250',
port=3306,
user='root',
password='********'
)
# 查看MySQL版本信息
print(conn.get_server_info()) # 5.5.27
# 获取到游标对象
cursor = conn.cursor()
# 选择数据库
conn.select_db("hanligang_data")
# 使用游标对象执行sql语句
sql = "select * from tstudent"
cursor.execute(sql)
# 获取查询结果
results: tuple = cursor.fetchall()
print(results) # 结果是一个大元组，里面包含了每一行数据的小元组
for r in results:
print(r)
conn.close() # 关闭
查询结果（里面身份证为for循环生产的虚拟数据）：
('0000000118', '魏欣若', '男', '6968390833530048', '1984-09-19 00:00:00.000', 'weixinruo@91xueit.com', '开发', '2019-12-18 17:58:52.85')
('0000000119', '胡辉香', '男', '1687819817472568', '1985-02-18 00:00:00.000', 'huhuixiang@91xueit.com', '开发', '2019-12-18 17:58:52.85')
('0000000120', '姜利维', '男', '1367699597056572', '1988-10-17 00:00:00.000', 'jiangliwei@91xueit.com', '网络', '2019-12-18 17:58:52.85')
('0000000121', '马文霞', '男', '2921194355462422', '1984-03-11 00:00:00.000', 'mawenxia@91xueit.com', '测试', '2019-12-18 17:58:52.85')
('0000000122', '于爽轮', '女', '3826182221042389', '1981-03-27 00:00:00.000', 'yushuanglun@91xueit.com', '网络', '2019-12-18 17:58:52.85')
('0000000123', '吴军雪', '女', '361896711952425', '1988-10-05 00:00:00.000', 'wujunxue@91xueit.com', '开发', '2019-12-18 17:58:52.85')
('0000000124', '廖菲以', '男', '6206075708967862', '1984-10-07 00:00:00.000', 'liaofeiyi@91xueit.com', '开发', '2019-12-18 17:58:52.85')
('0000000125', '孔国馨', '男', '6196758315299228', '1984-04-07 00:00:00.000', 'kongguoxin@91xueit.com', '开发', '2019-12-18 17:58:52.85')
('0000000126', '李伟伟', '女', '9973233711794250', '1988-08-05 00:00:00.000', 'liweiwei@91xueit.com', '开发', '2019-12-18 17:58:52.85')
import pymysql
db = pymysql.connect(
host='localhost', user='root', password='mysql', #连接到数据库需要指定的最少信息
db = 'demo', database='demo', #指定选择的数据库，任选一个都可，前面是简写
passwd='mysql', #密码也可以用passwd简写形式
autocommit=True, #设置修改自动提交到数据库
auth_plugin_map='mysql_native_password'
) #设置身份认证，8.0版本之后建议加上
cursor = db.cursor() #创建一个指针，之后基本所有的操作都是使用指针来实现
cursor.execute('show databases;') #执行SQL语句
db.commit() #将修改提交到数据库——如果连接时参数autocommit设置为false要加上
cursor.fetchall() #获取上一条SQL语句的执行结果，如查询结果等
cursor.fetchone() #获取执行结果的一行
db.close() #关闭数据库
为了让程序更加健壮，我们一般不会链接数据库这样调用该函数，先不说麻烦与否，如果我们经常要修改连接参数，那么我们所有调用地方都会修改，这样的代码十分不健壮。所以我们应该使用模块化进行解决，代码如下：
import pymysql
def connect_mysql():
db_config = {
'host':'127.0.0.1',
'port':3306,
'user':'ian',
'password':'ian123',
'charset':'utf8'
}
conn = pymysql.connect(**db_config)
return conn
查询函数
def query_students_fetchone():
sql = 'SELECT * FROM students'
rows = cursor.execute(sql)
print('There are %d students' % rows)
for i in range(rows):
student = cursor.fetchone()
print(student)
增删数据库
增删改，操作方式都一样，只是mysql语句的不同，返回的值是影响的行数。需要注意的是，我们需要提交事务，如果不提交事务，语句是不执行的，只是显示执行成功，但是表中并没有进行相关操作。代码如下：
import pymysql
def insert_mysql():
no = int(input('编号: '))
name = input('名称: ')
location = input('所在地: ')
# 1. 创建连接
conn = pymysql.connect(host='127.0.0.1', port=3306,
user='ian', password='ian123',
database='test', charset='utf8')
try:
# 2. 获取游标对象
with conn.cursor() as cursor:
# 3. 通过游标对象向数据库服务器发出SQL语句
affected_rows = cursor.execute(
'insert into `tb_test` values (%s, %s, %s)',
(no, name, location)
)
if affected_rows == 1:
print('新增成功!!!')
# 4. 提交事务
conn.commit()
except pymysql.MySQLError as err:
# 4. 回滚事务
conn.rollback()
print(type(err), err)
finally:
# 5. 关闭连接释放资源
conn.close()
def del_mysql():
no = int(input('编号: '))
# 1. 创建连接
conn = pymysql.connect(host='127.0.0.1', port=3306,
user='ian', password='ian123',
database='test', charset='utf8',
autocommit=True)
try:
# 2. 获取游标对象
with conn.cursor() as cursor:
# 3. 通过游标对象向数据库服务器发出SQL语句
affected_rows = cursor.execute(
'delete from `tb_dept` where `dno`=%s',
(no, )
)
if affected_rows == 1:
print('删除成功!!!')
finally:
# 5. 关闭连接释放资源
conn.close()
insert_mysql()
del_mysql()
封装mysql类
import pymysql
from pathlib import Path
class ConnectMsql:
def __init__(self, host='127.0.0.1', port=3306, user='ian',
password='ian123', database="test", filename: str = "test.sql"):
"""
:param host: 域名
:param port: 端口
:param user: 用户名
:param password: 密码
:param database: 数据库名
:param filename: 文件名称
"""
self._host: str = host
self._port: int = port
self._user: str = user
self._password: str = password
self._database: str = database
self._file_path = Path(__file__).parent.joinpath(filename)
def _show_databases_and_create(self):
"""
查询数据库是否存在，不存在则进行新建操作
:return:
"""
connection = pymysql.connect(host=self._host, port=self._port, user=self._user, password=self._password,
cursorclass=pymysql.cursors.DictCursor)
with connection:
with connection.cursor() as cursor:
cursor.execute('show databases;')
result = cursor.fetchall()
results = self._database not in tuple(x["Database"] for x in result)
if results:
with connection.cursor() as cursor:
cursor.execute(f'create database {self._database};')
with connection.cursor() as cursor:
cursor.execute('show databases;')
result = cursor.fetchall()
results = self._database in tuple(x["Database"] for x in result)
return results if results else result
else:
return True
def _export_databases_data(self):
"""
读取.sql文件，解析处理后，执行sql语句
:return:
"""
if self._show_databases_and_create() is True:
connection = pymysql.connect(host=self._host, port=self._port, user=self._user, password=self._password,
database=self._database, charset='utf8')
# 读取sql文件，并提取出sql语句
results, results_list = "", []
with open(self._file_path, mode="r+", encoding="utf-8") as r:
for sql in r.readlines():
# 去除数据中的“\n”和“\r”字符
sql = sql.replace("\n", "").replace("\r", "")
# 获取不是“--”开头且不是“--”结束的数据
if not sql.startswith("--") and not sql.endswith("--"):
# 获取不是“--”的数据
if not sql.startswith("--"):
results = results + sql
# 根据“;”分割数据，处理后插入列表中
for i in results.split(";"):
if i.startswith("/*"):
results_list.append(i.split("*/")[1] + ";")
else:
results_list.append(i + ";")
# 执行sql语句
with connection:
with connection.cursor() as cursor:
# 循环获取sql语句
for x in results_list[:-1]:
# 执行sql语句
cursor.execute(x)
# 提交事务
connection.commit()
else:
return "sql全部语句执行成功 ！"
@property
def sql_run(self):
"""
执行方法
:return:
"""
return self._export_databases_data()
if __name__ == '__main__':
res = ConnectMsql().sql_run
print(res)