零基础学Python之整合MySQL

news2024/9/24 1:20:20

Python 标准数据库接口为 Python DB-API，Python DB-API为开发人员提供了数据库应用编程接口。

不同的数据库你需要下载不同的DB API模块，例如你需要访问Oracle数据库和Mysql数据，你需要下载Oracle和MySQL数据库模块。

DB-API 是一个规范. 它定义了一系列必须的对象和数据库存取方式, 以便为各种各样的底层数据库系统和多种多样的数据库接口程序提供一致的访问接口。

Python的DB-API，为大多数的数据库实现了接口，使用它连接各数据库后，就可以用相同的方式操作各数据库。

Python DB-API使用流程：

引入 API 模块。
获取与数据库的连接。
执行SQL语句和存储过程。
关闭数据库连接。

Python中操作MySQL数据库有多种方法，例如使用mysql-connector-python、PyMySQL、mysqlclient等库。本文将以PyMySQL为例，介绍如何在Python中使用PyMySQL库进行MySQL数据库操作。

在Python中使用PyMySQL库进行MySQL数据库操作，需要先安装PyMySQL库。可以使用pip命令在命令行中安装PyMySQL库。

pip install pymysql

1.连接MySQL数据库

使用 connect() 方法连接到MySQL数据库。需要提供主机名、用户名、密码和数据库名。

import pymysql

connection = pymysql.connect(
            host="47.93.159.97",
            port=3306,
            user="root",
            password="mysql8test.",
            db="user",
            charset="utf8mb4"
        )

2.SQL语句执行

使用 execute() 方法执行对应的SQL语句。

import pymysql

# 获取到connection对象之后，编写sql调用execute()方法。
cursor = connection.cursor()
sql = "增删改查sql语句"
# 参数
date = ("元组类型",)
# 执行SQL
affect_row = cursor.execute(sql,date)

3.Python操作MySQL案例

了解完 connect() 、execute() 下面我们直接来一个案例演示下Python操作MySQL的增删改查。

需求：编写一个程序，操作数据库需要对人员进行增删改查。

（1）准备数据库表格user表

-- `user`.`user` definition

CREATE TABLE `user` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `name` varchar(100) DEFAULT NULL COMMENT '姓名',
  `age` varchar(100) DEFAULT NULL COMMENT '年龄',
  `sex` varchar(100) DEFAULT NULL COMMENT '性别',
  `address` varchar(100) DEFAULT NULL COMMENT '地址',
  `career` varchar(100) DEFAULT NULL COMMENT '职业',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

（2）封装数据库的工具类

import pymysql

'''
数据库操作工具类
'''

class DBUtil:

    # 创建连接对象
    @staticmethod
    def get_connection():
        connection = pymysql.connect(
            host="ip地址",
            port=3306,
            user="user",
            password="password",
            db="user",
            charset="utf8mb4"
        )
        return connection

    # 关闭连接对象
    @staticmethod
    def close(connection, cursor):
        # 更改提交
        connection.commit()
        # 关闭游标对象
        cursor.close()
        # 关闭连接对象
        connection.close()

（3）数据库增删改查方法编写

from DBUtil import DBUtil

class UserDao:

    @staticmethod
    def __addUser__(name, age, sex, address, career):
        connection = DBUtil.get_connection()
        cursor = connection.cursor()
        sql = "insert into user (name,age,sex,address,career) values (%s,%s,%s,%s,%s)"
        data = [(name,age,sex,address,career)]
        affect_row = cursor.executemany(sql,data)
        print("受影响的行数：",affect_row)
        if affect_row > 0:
            print("添加成功")
        else:
            print("添加失败")
        DBUtil.close(connection,cursor)

    @staticmethod
    def __updateUser__(user_id, name, age, sex, address, career):
        connection = DBUtil.get_connection()
        cursor = connection.cursor()
        sql = "update user set name = %s,age = %s,sex=%s,address=%s,career=%s where id = %s"
        date = (name,age,sex,address,career,user_id)
        affect_row = cursor.execute(sql,date)
        if affect_row > 0:
            print("修改成功")
        else:
            print("修改失败，当前人员不存在")
        DBUtil.close(connection,cursor)

    @staticmethod
    def __deleteUser__(user_id):
        connection = DBUtil.get_connection()
        cursor = connection.cursor()
        sql = "delete from user where id =%s"
        affect_row = cursor.execute(sql, (user_id,))
        if affect_row > 0:
            print("删除成功")
        else:
            print("删除失败，当前人员不存在")
        DBUtil.close(connection, cursor)

    @staticmethod
    def __selectUsers__():
        connection = DBUtil.get_connection()
        cursor = connection.cursor()
        sql = "select * from user"
        cursor.execute(sql)
        rows = cursor.fetchall()
        for row in rows:
            print("编号:",row[0],"姓名:",row[1],"年龄:",row[2],"性别:",row[3],"地址:",row[4],"职业:",row[5])
        DBUtil.close(connection,cursor)

（4）主程序入口编写

from UserDao import UserDao

def main():
    # 入口
    while True:
        print("[1]增加用户信息")
        print("[2]删除用户信息")
        print("[3]修改用户信息")
        print("[4]查询用户信息")
        print("[5]退出系统")
        choose = int(input("请输入你要执行的选项："))
        if choose == 1:
            name = input("姓名：")
            age = input("年龄：")
            sex = input("性别：")
            address = input("地址：")
            career = input("职业：")
            UserDao.__addUser__(name, age, sex, address, career)
        elif choose == 2:
            user_id = input("用户ID：")
            UserDao.__deleteUser__(user_id)
        elif choose == 3:
            user_id = input("用户ID：")
            name = input("姓名：")
            age = input("年龄：")
            sex = input("性别：")
            address = input("地址：")
            career = input("职业：")
            UserDao.__updateUser__(user_id, name, age, sex, address, career)
        elif choose == 4:
            UserDao.__selectUsers__()
        else:
            print("退出系统")
            break

if __name__ == "__main__":
    main()