Python连接MySQL数据库

安装MySQL

参考链接

安装Navicat Premium 16

参考链接

连接MySQL

安装库

pip install pymysql

MySQL封装

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @File    : MySqlUtils.py
# @Description :MySQL封装
# 方法：
# 1. mysql = MySqlUtils(操作的表名，参数)  连接数据库
#     mysql.operate_table_name = "tb_user"   修改要操作的表
# 2. insert((字段1,字段2,...),(字段1,字段2,...))    插入数据
#    insert_by_sql(sql语句)   自定义插入数据
# 3. delete_by_sql(sql语句)  自定义删除数据
# 4. update_by_sql(sql语句)   修改数据
# 5. select_many()  条件查询
# 6. select_all()   查询所有数据
# 7. select_by_sql() 自定义查询


import pymysql


class MySqlUtils:
    def __init__(self, operate_table_name: str, my_sqldb_config_param: dict):
        """
        获取MySQL对象\n
        :param operate_table_name: 要进行操作的表名
        :param my_sqldb_config_param:连接本机MySQL所需的配置参数
        例子：\n
        my_sqldb_config_param = {
            "host": "127.0.0.1", #连接主机的ip\n
            "port": 3306,   #连接主机的端口\n
            "user": "root", #本地数据库的用户名\n
            "password": "***", #本地数据库的密码\n
            "database": "pythondemo", #连接的数据库\n
            "charset": "utf8" #设置编码格式
        }
        """
        assert isinstance(my_sqldb_config_param, dict), "请以字典类型的格式传入！"
        self._operate_table_name = operate_table_name
        try:
            self._conn = pymysql.connect(**my_sqldb_config_param)  # 连接数据库，配置参数
            self._cursor = self._conn.cursor()  # 创建一个游标，用来执行查询
            self._get_field()  # 获取此表中的字段名
        except Exception as e:
            raise Exception(f"数据库连接失败！！！\n请检查表名、配置参数是否正确或检查本地数据库是否已启动！\n{e}")

    @property
    def get_connect(self):
        """
        获取_conn对象\n
        :return: _conn对象
        """
        return self._conn

    @property
    def get_cursor(self):
        """
        获取_cursor对象
        :return: _cursor对象
        """
        return self._cursor

    @property
    def get_description(self):
        """
        获取__desc对象
        :return: __desc对象
        """
        return self._desc

    @property
    def operate_table_name(self):
        """
        获取正在操作的表名
        :return: 正在操作的表名
        """
        return f"正在操作 {self._operate_table_name}表！！！"

    @operate_table_name.setter
    def operate_table_name(self, operate_table_name):
        """
        修改要操作的表\n
        :param operate_table_name: 要操作的表
        """
        assert operate_table_name != "", "请输入要操作的表名！"
        print(f"{self._operate_table_name} 表已被更换！")
        self._operate_table_name = operate_table_name
        self._get_field()

    def _get_field(self):
        """
        获取此表中的字段名\n
        :return: 执行结果
        """
        self._cursor.execute(f"select * from {self._operate_table_name}")
        self._desc = self._cursor.description
        self._field_ = []
        for field in self._desc:
            self._field_.append(field[0])

    def _sql(self, sql: str, msg=""):
        """
        执行sql语句\n
        :param sql: sql语句
        :param msg:
        :return: 执行结果，bool类型
        """
        try:
            self._cursor.execute(sql)  # 执行sql语句
            self._conn.commit()  # 执行sql语句后，进行提交
            if msg: print(f"数据{msg}成功！")
            return True
        except Exception as e:
            if msg: print(f"\033[31m数据{msg}失败！！！\n{e} \033[0m")
            self._conn.rollback()  # 执行sql语句失败，进行回滚
            return False

    def insert(self, *value):
        """
        插入数据
        :param value: 插入数据
        """
        if not isinstance(value[0], tuple): raise Exception("要求传入的参数类型为tuple元组！！！")
        if len(value) == 1:
            value = value[0]
        else:
            value = str(value)[1:-1]
        sql = f"insert into {self._operate_table_name}({','.join(self._field_[1:])}) values {value}"
        if not self._sql(sql, f"{value}插入"):
            print("\n\033[31m：请检查每一条记录字段是否正确！！！\033[0m\n")

    def insert_by_sql(self, sql:str):
        """
        插入：自定义sql语句插入数据\n
        :param sql: sql语句
        :return: 执行结果
        """
        return self._sql(sql, "插入")

    def delete_by_sql(self, sql:str):
        """
        删除：自定义sql语句删除数据\n
        :param sql: sql语句
        :return: 执行结果
        """
        self._sql(sql, "删除")

    def update_by_sql(self, sql:str):
        """
        修改：自定义sql语句修改数据\n
        :param sql: sql语句
        :return: 执行结果
        """
        self._sql(sql, "修改")

    def select_many(self, num: int, query_builder: dict = None, field="*"):
        """
        查询：指定查询多少条数数据,可根据简单条件查询（where 字段=”“）\n
        :param num: 要查询的数据数目
        :param query_builder:查询条件，字典类型
        :param field: 列名
        :return:
        """
        if field != "*": field = ','.join(field)
        sql = f"select {field} from {self._operate_table_name}"
        if query_builder:
            if isinstance(query_builder, dict) and len(query_builder) == 1:
                query_builder = list(query_builder.items())[0]
                sql = f"select {field} from {self._operate_table_name} where {query_builder[0]}='{query_builder[1]}'"
            else:
                raise Exception("要求输入的条件为dict(字典)类型并且只能有一对键值（：len(dict）=1）！！！")
        self._cursor.execute(sql)
        return self._cursor.fetchmany(num)

    def select_all(self, field="*"):
        """
        查询：所有数据\n
        :param field: 列名
        :return: 执行结果
        """
        if field != "*": field = ','.join(field)
        sql = f"select {field} from {self._operate_table_name}"
        self._cursor.execute(sql)
        return self._cursor.fetchall()

    def select_by_sql(self, sql:str):
        """
        查询：自定义sql语句查询数据\n
        :param sql: sql数据
        :return: 执行结果
        """
        try:
            self._cursor.execute(sql)
            return self._cursor.fetchall()
        except Exception as e:
            print(f"\033[31m：数据查询失败！！！\n{e} \033[0m")

    def __del__(self):
        """
        当对象被销毁时，游标先关闭,连接后关闭
        """
        self._cursor.close()
        self._conn.close()