Python 测开27期 - WL - 学习笔记 - 常用第三方库 pymysql

学习笔记
,
1

常用第三方库 pymysql

pymysql 概述

  • Python 的数据库接口标准是 Python DB-API
  • PyMySQL 是从 Python 连接到 MySQL 数据库服务器的接口
  • PyMySQL 的目标是成为 MySQLdb 的替代品
  • 官方文档：https://pymysql.readthedocs.io/

pymysql 安装

  • 使用 pip 安装：pip install pymysql
  • 使用 Pycharm 界面安装

pymysql 连接数据库

  • host：MySQL 服务器地址
  • user：用户名
  • password：密码
  • database：数据库名称
  • charset：编码方式，推荐使用 utf8mb4
  • 完成后必须关闭数据库连接
    image 
import pymysql

# 简历数据库链接
conn = pymysql.connect(host='106.53.104.31',
                       user='root',
                       password='123456',
                       database='demo',
                       charset='utf8mb4')
# 关闭数据库链接
conn.close()

封装获取连接的函数

image

pymysql 入门实例

  • 获取连接对象
    ** 打开
    ** 关闭
  • 获取游标对象
    ** 执行SQL
    ** 查询记录
    image 
from .sql_conf import get_conn

def test_demo():
    # 打开连接
    conn = get_conn()
    # 获取游标
    couser = conn.cursor()
    # 执行SQL
    couser.execute('SELECT VERSION();')
    # 获取单挑结果
    version = couser.fetchone()
    print(version)
    # 关闭连接
    conn.close()

pymysql 数据库操作

CRUD 操作

创建表

image
image827×462 72.2 KB 

from . import get_conn

def test_demo():
    # 打开连接
    conn = get_conn()
    # 获取游标
    couser = conn.cursor()
    # SQL 语句
    sql = """
    CREATE TABLE 'testcase'(
    'id' int(11) NOT NULL AUTO_INCHEMENT,
    'title' varchar(255) COLLATE utf8_bin NOT NULL,
    'expect' varchar(255) COLLATE utf8_bin NOT NULL,
    'owner' varchar(255) COLLATE utf8_bin NOT NULL,
    PRIMARY KEY('id')
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; 
    """
    # 执行SQL
    couser.execute(sql)
    # 关闭连接
    conn.close()

插入记录

image

查询记录

方法 注释
fetchone() 获取单条记录
fetchmant(n) 获取 n 条记录
fetchall() 获取所有结果记录

image

from SQL.sql_conf import get_conn
import sys

def test_select():
    conn = get_conn()
    curser = conn.cursor()
    sql = """SELECT * FROM testcase;"""
    # 捕获异常
    try:
        # 执行sql
        curser.execute(sql)
        # 获取查询一条结果
        record = curser.fetchone()
        # 获取查询一3条结果
        record2 = curser.fetchmany(3)
        # 获取所有查询结果
        record3 = curser.fetchall()
        return {record}, {record2}, {record3}
    except Exception as e:
        # 打印执行的错误信息
        print(sys.exc_info())
    finally:
        # 关闭数据库连接
        conn.close()

更新记录

image
image826×388 40.7 KB 

from SQL.sql_conf import get_conn

def test_update():
    conn = get_conn()
    curser = conn.cursor()
    sql = """UPDATE testcase SET id=4,title='S11全球总决赛',expect='冠军',owner='EDG';"""
    try:
        # 执行sql
        curser.execute(sql)
        # 提交事务
        conn.commit()
    except:
        # 事务回滚
        conn.rollback()
    finally:
        # 关闭数据库连接
        conn.close()

删除记录

image

from SQL.sql_conf import get_conn

def test_delete():
    conn = get_conn()
    curser = conn.cursor()
    sql = """DEKETE FROM  testcase WHERE id=4;"""
    try:
        # 执行sql
        curser.execute(sql)
        # 提交事务
        conn.commit()
    except:
        # 事务回滚
        conn.rollback()
    finally:
        # 关闭数据库连接
        conn.close()

执行事务

  • 提交：commit
  • 回滚：rollback
  • try-catch-finally
    image 
from SQL.sql_conf import get_conn


def test_inset():
    conn = get_conn()
    curser = conn.cursor()
    sql = """INSERT INTO testcase (id,title,expect,owner) VALUES (1,'S11全球总决赛','冠军','EDG');"""
    try:
        # 执行sql
        curser.execute(sql)
        # 提交事务
        conn.commit()
    except:
        # 事务回滚
        conn.rollback()
    finally:
        # 关闭数据库连接
        conn.close()