Python脚本之连接MySQL【四】

news2024/11/26 20:48:49

本文为博主原创，未经授权，严禁转载及使用。
本文链接：https://blog.csdn.net/zyooooxie/article/details/124640412

之前写了篇 Python脚本之连接MySQL【三】，日常使用过程中，代码实际有很多改动，特此更新；

【实际这篇博客推迟发布N个月】

个人博客：https://blog.csdn.net/zyooooxie

【以下所有内容仅为个人项目经历，如有不同，纯属正常】

Cursor类的nextset()、_nextset()

https://peps.python.org/pep-0249/#nextset

Cursor.nextset()

This method will make the cursor skip to the next available set, discarding any remaining rows from the current set.

If there are no more sets, the method returns None. Otherwise, it returns a true value and subsequent calls to the .fetch*() methods will return rows from the next result set.

在这里插入图片描述

方法优化【2】

【读取ini配置文件 + 建立、关闭数据库连接】

@filename: db.ini

[zyooooxie_db]
user = user_zy
host = host_zy
passwd = passwd_zy
database = database_zy

"""
@blog: https://blog.csdn.net/zyooooxie
@qq: 153132336
@email: zyooooxie@gmail.com
"""

from pymysql.connections import Connection
from pymysql.cursors import Cursor
from pymysql.constants import CLIENT

def read_ini(ini_path: str = 'db.ini'):
    """
    读取配置文件
    :param ini_path:默认是数据库的配置文件
    :return:
    """

    if os.path.isfile(ini_path):
        config = ConfigParser()
        config.read(ini_path, encoding='UTF-8')

        return {cf: config.items(cf) for cf in config.sections()}


def connect_db(db_name: str) -> (Connection, Cursor):
    """
    建立链接，传参为 具体的db
    :param db_name:
    :return:
    """
    ini_dict = read_ini()

    if db_name in ini_dict:
        res_list = ini_dict.get(db_name)
        res_dict = dict(res_list)

    else:
        raise Exception('传参不合法')

    db = pymysql.connect(port=3306, charset='utf8', autocommit=True, client_flag=CLIENT.MULTI_STATEMENTS,
                         use_unicode=True, **res_dict)  # client_flag传值：默认可以同时执行多条sql语句的
    cur = db.cursor()

    Log.debug('{} connect'.format(db_name))

    return db, cur


def close_db(db: Connection, cur: Cursor):
    """
    断开连接
    :param db:
    :param cur:
    :return:
    """
    cur.close()
    db.close()
    Log.debug('connect close')

exe_sql()、fetch_sql()、fetch_sqls()

"""
@blog: https://blog.csdn.net/zyooooxie
@qq: 153132336
@email: zyooooxie@gmail.com
"""

def exe_sqls(sql: str, db_name: str, db: Connection = None, cur: Cursor = None):
    """
    多条sql语句-execute()
    :param sql:
    :param db_name:
    :param db:
    :param cur:
    :return:
    """
    exe_sql(sql=sql, db_name=db_name, db=db, cur=cur)


def exe_sql(sql: str, db_name: str, exe_mode: str = None, data_list: list or tuple = None,
            db: Connection = None, cur: Cursor = None):
    """
    1条sql语句-execute()、executemany()
    :param sql:
    :param db_name:
    :param exe_mode:
    :param data_list:
    :param db:
    :param cur:
    :return:
    """

    if not bool(cur):
        db_use, cur_use = connect_db(db_name=db_name)

    else:
        db_use, cur_use = db, cur

    Log.info(sql)

    if data_list is None and exe_mode is None:
        try_str = """cur_use.execute(sql)"""

    elif exe_mode == 'execute' and data_list is not None:
        assert sql.find('(%s') != -1
        try_str = """cur_use.execute(sql, data_list)"""

    elif exe_mode == 'executemany' and data_list is not None:
        assert sql.find('(%s') != -1
        try_str = """cur_use.executemany(sql, data_list)"""

    else:
        Log.error('{}--{}'.format(exe_mode, data_list))
        raise Exception('Execute Error')

    try:
        result = eval(try_str, locals())

    except Exception as e:
        db_use.rollback()
        Log.error(e.args)
        Log.info(traceback.format_exc())

        result = False
        Log.error('sql执行有问题')

    else:
        execute_sql_result_check(result)

    finally:

        if not bool(cur):
            close_db(db=db_use, cur=cur_use)


def execute_sql_result_check(result):
    """
    SQL Execute结果检查
    :param result:
    :return:
    """

    if not result:
        Log.error('{}-Number of affected rows'.format(result))

    else:
        Log.debug('Execute Succeed:{}'.format(result))


def fetch_sqls(sql: str, db_name: str, db: Connection = None, cur: Cursor = None) -> List[tuple]:
    """
    多条sql语句-fetchall()
    :param sql:
    :param db_name:
    :param db:
    :param cur:
    :return:
    """
    if not bool(cur):
        db_use, cur_use = connect_db(db_name=db_name)

    else:
        db_use, cur_use = db, cur

    Log.info(sql)

    try:
        data = list()

        cur_use.execute(sql)
        data.append(cur_use.fetchall())

        # while True:
        #
        #     if cur_use.nextset():
        #         data.append(cur_use.fetchall())
        #
        #     else:
        #         break

        while cur_use.nextset():
            data.append(cur_use.fetchall())

    except Exception as e:
        db_use.rollback()

        Log.debug(e.args)
        Log.info(traceback.format_exc())

        data = False
        Log.error('sql执行有问题')

    else:

        for d in data:
            fetch_sql_result_check(d)

    finally:

        if not bool(cur):
            close_db(db=db_use, cur=cur_use)

    return data


def fetch_sql(sql: str, db_name: str, fetch_mode: str = 'fetchall', db: Connection = None, cur: Cursor = None):
    """
    1条sql语句-fetchone()、fetchall()
    :param sql:
    :param db_name:
    :param fetch_mode:
    :param db:
    :param cur:
    :return:
    """

    if not bool(cur):
        db_use, cur_use = connect_db(db_name=db_name)

    else:
        db_use, cur_use = db, cur

    Log.info(sql)

    if fetch_mode == 'fetchall':
        try_str = """cur_use.fetchall()"""

    elif fetch_mode == 'fetchone':  # 很少用到
        try_str = """cur_use.fetchone()"""

    else:
        Log.error('fetch_mode: {}'.format(fetch_mode))
        raise Exception(fetch_mode)

    try:
        cur_use.execute(sql)
        data = eval(try_str, locals())

    except Exception as e:
        db_use.rollback()

        Log.debug(e.args)
        Log.info(traceback.format_exc())

        data = False
        Log.error('sql执行有问题')

    else:
        fetch_sql_result_check(data)

    finally:

        if not bool(cur):
            close_db(db=db_use, cur=cur_use)

    return data


def fetch_sql_result_check(result):
    """
    SQL Fetch结果检查
    :param result:
    :return:
    """

    if not result:
        Log.error('{}-Fetch Nothing'.format(result))

    else:
        Log.debug('Fetch Succeed')

实际应用【2】

"""
@blog: https://blog.csdn.net/zyooooxie
@qq: 153132336
@email: zyooooxie@gmail.com
"""

def test_0801():
    from XXX_use.common_mysql import fetch_sql, connect_db, close_db, fetch_sqls, exe_sql, exe_sqls

    db_, cur_ = connect_db(db_name='zy_db')

    sql = """
    SELECT  *  FROM `table_c_r`    ORDER BY `update_date` DESC ;
    """

    data = fetch_sql(sql=sql, db_name='zy_db', fetch_mode='fetchall', db=db_, cur=cur_)
    Log.info(data)

    data = fetch_sql(sql=sql, db_name='zy_db', fetch_mode='fetchone', db=db_, cur=cur_)
    Log.info(data)

    data = fetch_sql(sql=sql, db_name='zy_db', db=db_, cur=cur_)
    Log.info(data)

    sql = """
    SELECT  *  FROM `table_c_r_g_m_s`    ORDER BY `update_date` DESC LIMIT 50 ;
    
    # 这条sql查不到数据
    SELECT  *  FROM `table_c_b_r_r`  WHERE   `delete_flag` = 1   ORDER BY `update_date` DESC  ; 
    
    SELECT  *  FROM `table_c_r`    ORDER BY `update_date` DESC LIMIT 1;
    """

    res = fetch_sqls(sql=sql, db_name='zy_db', db=db_, cur=cur_)
    Log.info(res)

    data_list = [('TEST0801' + 'ZYOOOOXIE_' + str(i), 'Name' + str(i), 'zyooooxie') for i in
                 random.sample(range(500), 10)]
    Log.info(data_list)

    sql = """
    insert into `table_c_g_c`     (`c_i`, `name`, `owner`) 
    VALUES   (%s, %s, %s);
    """
    exe_sql(sql=sql, db_name='zy_db', exe_mode='executemany', data_list=data_list, db=db_, cur=cur_)

    exe_sql(sql=sql, db_name='zy_db', exe_mode='execute', data_list=data_list[-1], db=db_, cur=cur_)

    sql = """
    insert into `table_c_g_c`     (`c_i`, `name`, `owner`) 
    VALUES     ('TEST0801ZYOOOOXIE_11','Name_11', 'zyooooxie') ; 

    insert into table_c_y_a    (username, password)
    values    ('TEST0801_11', 'pwd_11');

    insert into `table_c_g_c`    (`c_i`, `name`, `owner`) 
    VALUES     ('TEST0801ZYOOOOXIE_12','Name_12', 'zyooooxie') ;    
    """
    exe_sql(sql=sql, db_name='zy_db', db=db_, cur=cur_)
    exe_sqls(sql=sql, db_name='zy_db', db=db_, cur=cur_)

    sql_ = """

    DELETE FROM `table_c_g_c` WHERE c_i LIKE 'TEST0801%' ;
    DELETE FROM `table_c_y_a` WHERE username LIKE 'TEST0801%' ;
    DELETE FROM `table_c_y_a` WHERE username LIKE 'TEST080808%' ;

    """
    exe_sql(sql=sql_, db_name='zy_db', db=db_, cur=cur_)
    exe_sqls(sql=sql_, db_name='zy_db', db=db_, cur=cur_)

    close_db(db_, cur_)

本文链接：https://blog.csdn.net/zyooooxie/article/details/124640412

个人博客 https://blog.csdn.net/zyooooxie

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/877702.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

数据分析--帆软报表--大数据大屏

数据分析--帆软报表--大数据大屏

进入国企公司学习有一段时间了，岗位是数据分析方向------ 母前使用的是帆软工具进行的开发。 可以进行大数据大屏 也可使嵌入到手机端。 下面是例子
阅读更多...
IntelliJ IDEA快捷键大全 + 动图演示

IntelliJ IDEA快捷键大全 + 动图演示

IntelliJ IDEA快捷键大全 动图演示 😍 一、构建/编译Ctrl F9：构建项目Ctrl Shift F9：重新编译当前类 😍 二、文本编辑Ctrl X：剪切Ctrl C：复制Ctrl V：粘贴Ctrl Alt Shift V&#xff1a…
阅读更多...
Nginx安装及Minio集群反向动态代理配置(二)

Nginx安装及Minio集群反向动态代理配置(二)

安装所需插件 1、安装gcc gcc是linux下的编译器在此不多做解释，感兴趣的小伙伴可以去查一下相关资料，它可以编译 C,C,Ada,Object C和Java等语言 命令：查看gcc版本 [rootwww ~]# gcc -v -bash: gcc: 未找到命令 说明没有安装: 那就直接yu…
阅读更多...
mclust学习总结

mclust学习总结

mclust example1 import numpy as np import pandas as pd from matplotlib import pyplot as plt from sklearn.datasets import make_blobs X, y make_blobs(n_samples300, centers4, cluster_std0.60, random_state0) # n_features2是默认的 plt.scatter(X[:,0], X[:,1]) …
阅读更多...
Java【数据结构】二分查找

Java【数据结构】二分查找

&#x1f31e; 题目&#xff1a; &#x1f30f;在有序数组A中&#xff0c;查找目标值target &#x1f30f;如果找到返回索引 &#x1f30f;如果找不到返回-1 算法描述解释前提给定一个内含n个元素的有序数组A&#xff0c;满足A0<A1<A2<<An-1,一个待查值target1设…
阅读更多...
如何利用 EMC 模型解决能源服务提供商的瓶颈

如何利用 EMC 模型解决能源服务提供商的瓶颈

01. 什么是合同能源管理&#xff1f; 合同能源管理(EMC-Energy Management Contract) 是一种新型的市场化节能机制,其实质就是以减少的能源费用来支付节能项目全部成本的节能投资方式。&#xff1a;节能服务公司与用能单位以契约形式约定节能项目的节能目标&#xff0c;节能服…
阅读更多...
在 ubuntu 18.04 上使用源码升级 OpenSSH_7.6p1到 OpenSSH_9.3p1

在 ubuntu 18.04 上使用源码升级 OpenSSH_7.6p1到 OpenSSH_9.3p1

1、检查系统已安装的当前 SSH 版本 使用命令 ssh -V 查看当前 ssh 版本&#xff0c;输出如下&#xff1a; OpenSSH_7.6p1 Ubuntu-4ubuntu0.7, OpenSSL 1.0.2n 7 Dec 20172、安装依赖&#xff0c;依次执行以下命令 sudo apt update sudo apt install build-essential zlib1g…
阅读更多...
springBoot 简单的demo

springBoot 简单的demo

springBoot 学习开始 场景开发流程1、创建项目2、导入依赖3、创建启动springBoot 项目的主入口程序4、创建业务程序5、在MainApplication文件运行程序6、将文件打包成jar包 遇到的问题未解决 希望大哥们帮忙--本地运行jar包报错 场景 浏览器发送hello请求&#xff0c;返回“he…
阅读更多...
培训报名小程序-用户注册

培训报名小程序-用户注册

目录 1 创建数据源2 注册用户3 判断用户是否注册4 完整代码总结 我们的培训报名小程序&#xff0c;用户每次打开时都需要填写个人信息才可以报名&#xff0c;如果用户多次报名课程&#xff0c;每次都需要填写个人信息&#xff0c;比较麻烦。 本篇我们就优化一下功能&#xff0c…
阅读更多...
[静态时序分析简明教程(十)]模式分析与约束管理

[静态时序分析简明教程(十)]模式分析与约束管理

静态时序分析简明教程-组合电路路径 一、写在前面1.1 快速导航链接 二、模式分析三、约束管理3.1 自顶向下的方法3.2 自底向上的方法 四、总结 一、写在前面 一个数字芯片工程师的核心竞争力是什么&#xff1f;不同的工程师可能给出不同的答复&#xff0c;有些人可能提到硬件描…
阅读更多...
新品牌怎么进行产品营销，小红书布局指南!

新品牌怎么进行产品营销，小红书布局指南!

随着互联网的快速发展&#xff0c;新品牌们逐渐意识到小红书这一平台的潜力。平台庞大的用户群体和高度活跃的内容创作者&#xff0c;成为新品牌布局的理想之地。今天&#xff0c;就来和大家分享一下新品牌怎么进行产品营销&#xff0c;小红书布局指南&#xff01; 第一步&…
阅读更多...
ARM 作业1

ARM 作业1

一、思维导图 二、 1. 2. .text 文本段 .globl _start 声明_start:mov r0,#0mov r1,#0fun:cmp r1,#100bhi stopadd r0,r0,r1add r1,r1,#1b fun stop:b stop .end
阅读更多...
简单谈谈 EMP-SSL：自监督对比学习的一种极简主义风

简单谈谈 EMP-SSL：自监督对比学习的一种极简主义风

论文链接&#xff1a;https://arxiv.org/pdf/2304.03977.pdf 代码&#xff1a;https://github.com/tsb0601/EMP-SSL 其他学习链接&#xff1a;突破自监督学习效率极限&#xff01;马毅、LeCun联合发布EMP-SSL&#xff1a;无需花哨trick&#xff0c;30个epoch即可实现SOTA 主要…
阅读更多...
图解结构体大小和位域例子

图解结构体大小和位域例子

struct A {short a; char b; int c : 1; char d : 4; short e : 7; }; 备注&#xff1a;蓝色&#xff1a;表示占一个符号位空间红色&#xff1a;表示补齐其他颜色&#xff1a;实际最大值所占空间 &#xff08;1&#xff09;图解例1 st…
阅读更多...
ASPICE学习笔记

ASPICE学习笔记

文章目录 1. ASPICE是什么？2. ASPICE能干什么？2.1 过程参考模型2.2 过程评估模型参考1. ASPICE是什么？ ASPICE的全称是Automotive SPICE。很明显的看出ASPICE是由SPICE发展而来。而SPICE是由国际标准化组织ISO、国际电工委员会IEC、信息技术委员会JTC1发起制定的ISO15504标…
阅读更多...
VSCODE[配置ssh免密远程登录]

VSCODE[配置ssh免密远程登录]

配置ssh免密远程登录 本文摘录于&#xff1a;https://blog.csdn.net/qq_44571245/article/details/123031276只是做学习备份之用&#xff0c;绝无抄袭之意&#xff0c;有疑惑请联系本人&#xff01; 这里要注意如下几个地方: 1.要进入.ssh目录创建文件: 2.是拷贝带"ssh-…
阅读更多...
Android面试官：“来给我讲讲View绘制？”

Android面试官：“来给我讲讲View绘制？”

前言 迎面走来的一位中年男子&#xff0c;他一手拿着保温杯&#xff0c;一手抱着笔记本电脑&#xff0c;顶着惺忪的睡眼&#xff0c;不紧不慢地走着&#xff0c;不多的几根头发在他头顶自由飞翔。过了一会&#xff0c;他面对着我坐下&#xff0c;放下电脑和保温杯&#xff0c;…
阅读更多...
【腾讯云Cloud Studio实战训练营】用Vue+Vite快速构建完成律师H5页面

【腾讯云Cloud Studio实战训练营】用Vue+Vite快速构建完成律师H5页面

&#x1f440;前置了解&#xff1a;(官网 https://cloudstudio.net/) 什么是Cloud Studio&#xff1f; Cloud Studio 是基于浏览器的集成式开发环境&#xff08;IDE&#xff09;&#xff0c;为开发者提供了一个永不间断的云端工作站。用户在使用 Cloud Studio 时无需安装&#…
阅读更多...
计算机设计大赛国赛一等奖项目分享——基于多端融合的化工安全生产监管可视化系统

计算机设计大赛国赛一等奖项目分享——基于多端融合的化工安全生产监管可视化系统

文章目录 一、计算机设计大赛国赛一等奖二、项目背景三、项目简介四、系统架构五、系统功能结构六、项目特色&#xff08;1&#xff09;多端融合&#xff08;2&#xff09;数据可视化&#xff08;3&#xff09;计算机视觉&#xff08;目标检测&#xff09; 七、系统界面设计&am…
阅读更多...
源码断点分析Spring的占位符（Placeholder）是怎么工作的

源码断点分析Spring的占位符（Placeholder）是怎么工作的

项目中经常需要使用到占位符来满足多环境不同配置信息的需求&#xff0c;比如&#xff1a; <?xml version"1.0" encoding"UTF-8"?> <beans xmlns:xsi"http://www.w3.org/2001/XMLSchema-instance"xmlns"http://www.springframe…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023