python-mysql协程并发常用操作封装

目录

  • 前言
  • 封装代码
  • 测试代码
  • 参考


前言

协程异步操作MYSQL是常用的,博主这里在GitHub上找了两个包,databases和aiomysql,第一个包除了mysql外还支持其他的数据库,且操作MYSQL时底层也是使用的aiomysql,但文档内容比较少,所以选择了对aiomysql进行封装。使用

pip3 install aiomysql

安装即可。

封装代码

async_mysql.py

# coding:utf-8
import aiomysql
import traceback


class AsyncMySQLClient:
    def __init__(self, host: str, user: str, password: str, db: str, port: int = 3306, loop=None):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.db = db
        self.loop = loop
        self.pool = None

    # 创建连接池
    async def connect(self):
        try:
            self.pool = await aiomysql.create_pool(
                host=self.host,
                port=self.port,
                user=self.user,
                password=self.password,
                db=self.db,
                loop=self.loop,
                autocommit=True
            )
        except:
            print(f"connect error:{traceback.format_exc()}")

    # 关闭连接池
    async def close(self):
        self.pool.close()
        await self.pool.wait_closed()

    # 执行一条非查询类语句
    async def execute(self, query:str, args:tuple=None) -> int:
        async with self.pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute(query, args)
                return cur.rowcount

    # 批量执行非查询类语句
    async def executemany(self, query:str, args:list=None) -> int:
        async with self.pool.acquire() as conn:
            async with conn.cursor() as cur:
                await cur.executemany(query, args)
                return cur.rowcount

    # 查询单条数据
    async def fetchone(self, query:str, args:tuple=None) -> dict:
        async with self.pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                await cur.execute(query, args)
                return await cur.fetchone()

    # 查询多条数据
    async def fetchall(self, query:str, args=None) -> list:
        async with self.pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                await cur.execute(query, args)
                return await cur.fetchall()

    # 封装单条插入
    async def insert(self, table: str, data: dict) -> int:
        """
        :param table: 表名
        :param data: 数据
        :return:
        """
        keys = ','.join(data.keys())
        values = ','.join(['%s'] * len(data))
        query = f'INSERT INTO {table} ({keys}) VALUES ({values})'
        try:
            return await self.execute(query, tuple(data.values()))
        except:
            print(f"execute {query} with {data} failed, error{traceback.format_exc()}")
            return 0

    # 封装批量插入
    async def insert_many(self, table:str, data_list:list)->int:
        """
        :param table: 表名
        :param data_list: 数据列表
        :return:
        """
        keys = ','.join(data_list[0].keys())
        values = ','.join(['%s'] * len(data_list[0]))
        query = f'INSERT INTO {table} ({keys}) VALUES ({values})'
        args = [tuple(data.values()) for data in data_list]
        try:
            return await self.executemany(query, args)
        except:
            print(f"execute {query} with {args} failed, error{traceback.format_exc()}")
            return 0

非查询类的返回影响行数,查询类的返回数据,使用字典(或字典为元素的列表)。对于常用的插入和查询进行了封装,其他的使用execute函数即可。
使用时生成类的对象,之后connect,执行语句后close即可。

测试代码

import asyncio
import uuid
from async_mysql import AsyncMySQLClient


async def main():
   cli =  AsyncMySQLClient(host="127.0.0.1",user="root",password="12345678",db="test")
   await cli.connect()
   nums = await cli.insert(table="users",data={
       "id": uuid.uuid4(),
       "name": "lady_killer9"
   })
   print(f"插入{nums}条数据")
   random_id = uuid.uuid4()
   nums = await cli.insert_many(table="users",data_list=[{
       "id": random_id,
       "name": "lady_killer9"
   }])
   print(f"插入{nums}条数据")
   user = await cli.fetchone(query="select * from users where id = %s",args=(random_id,))
   print(f"查询到用户:{user}")
   users = await cli.fetchall(query="select * from users where name = 'lady_killer9'")
   print(f"所有用户:{users}")
   num = await cli.execute(query="update users set name='lady_killer8' where id = %s",args=(random_id,))
   if num == 1:
       print("修改成功")
   nums = await cli.execute(query="delete from users")
   print(f"删除{nums}条数据")
   await cli.close()

if __name__ == '__main__':
    asyncio.run(main())

截图如下:
在这里插入图片描述

参考

databases
aiomysql

更多python相关内容,参考:【python总结】python学习框架梳理

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

【大数据】Flink 内存管理(三):TaskManager 内存分配(理论篇)

Flink 内存管理(三):TaskManager 内存分配 1.配置 Total Memory2.配置 Heap and Managed Memory2.1 Task (Operator) Heap Memory2.2 Managed Memory 3.配置 Off-Heap Memory(Direct or Native)4.详细内存模型5.Framew…

YOLO系列论文阅读(v1--v3)

搞目标检测,绕不开的一个框架就是yolo,而且更糟糕的是,随着yolo的发展迭代,yolo网络可以做的事越来越多,语义分割,关键点检测,3D目标检测。。。这几天决定把YOLO系列彻底梳理一下,在…

C++的STL常用算法->常用遍历算法、常用查找算法、常用排序算法、常用拷贝和替换算法、常用算术生成算法、常用集合算法

#include<iostream> using namespace std; #include <algorithm> #include <vector> //常用遍历算法 for_each //普通函数 void print01(int val) { cout << val << " "; } //仿函数 //函数对象 class print02 { public: v…

Wireshark TS | Linux 系统对时问题

问题描述 节前业务运维同事提交了一个 case &#xff0c;说是部署在新业务区域的 Linux 服务器和老业务区域的 Linux 服务器无法对时&#xff0c;脚本里使用的是 clockdiff 命令&#xff0c;无法正常返回结果&#xff0c;而在老业务区域两台服务器之间执行命令就正常&#xff…

Java基于微信小程序的校园二手物品交易系统,附源码

博主介绍&#xff1a;✌程序员徐师兄、7年大厂程序员经历。全网粉丝12w、csdn博客专家、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ &#x1f345;文末获取源码联系&#x1f345; &#x1f447;&#x1f3fb; 精彩专栏推荐订阅&#x1f447;…

公厕智慧化_智慧化的公厕

公厕智慧化是现代城市建设中的重要一环。通过信息化、数字化和智慧化技术手段&#xff0c;实现对公共厕所的高效管理和服务&#xff0c;不仅提升了城市环境质量&#xff0c;还改善了居民生活品质。智慧公厕的智慧化包括监测、管理、服务和设备的智慧化&#xff0c;利用先进科技…

Unity中URP实现水体效果(水的深度)

文章目录 前言一、搭建预备场景1、新建一个面片&#xff0c;使其倾斜一个角度&#xff0c;来模拟水底和岸边的效果2、随便创建几个物体&#xff0c;作为与水面接触的物体3、再新建一个面片&#xff0c;作为水面 二、开始编写水体的Shader效果1、新建一个URP基础Shader2、把水体…

汇编语言movs指令学习

字符串传送指令(Move String Instruction) movs 该指令是把指针DS:SI所指向的字节、字或双字传送给指针ES:DI所指向内存单元&#xff0c;并根据标志位DF对寄存器DI和SI作相应增减。该指令的执行不影响任何标志位。 记不清这指令是8086就有的&#xff0c;还是386以后新加的&…

【Redis】常见的5种数据类型(上)

文章目录 1 :peach:前言:peach:2 :peach:Redis 基本的全局命令:peach:2.1 :apple:keys:apple:2.2 :apple:exists:apple:2.3 :apple:del:apple:2.4 :apple:expire:apple:2.5 :apple:ttl:apple:2.6 :apple:type:apple: 3 :peach:单线程架构:peach:4 :peach:Redis 的 5 种常见数据…

Qt_纯虚函数的信号和槽

简介 在C中&#xff0c;纯虚函数是一个在基类中声明但没有实现的虚函数。纯虚函数的声明以 “ 0” 结尾。纯虚函数的目的是为了提供一个接口&#xff0c;但是不提供实现。派生类必须实现纯虚函数&#xff0c;否则它也会成为一个抽象类。纯虚函数可以在基类中定义&#xff0c;也…

MySQL--索引结构

索引-索引结构 1. 概述2. 二叉树3. B-Tree4. BTree5. Hash 1. 概述 MySQL的索引是在存储引擎层实现的&#xff0c;不同的存储引擎有不同的索引结构&#xff0c;主要包含以下几种&#xff1a; 上述是MySQL中所支持的所有的索引结构&#xff0c;下面展示不同的存储引擎对于索引…

力扣382.链表随机节点

Problem: 382. 链表随机节点 文章目录 题目描述思路复杂度Code 题目描述 思路 由水塘抽样易得&#xff0c;当遇到i个元素&#xff0c;有 1 / i 1/i 1/i的概率选择该元素&#xff1b;则在实际操作中我们定义一个下标i从1开始遍历每次判断rand() % i 0&#xff08;该操作就是判断…

Chrome插件(二)—Hello World!

本小节将指导你从头到尾创建一个基本的Chrome插件&#xff0c;你可以认为是chrome插件开发的“hello world”&#xff01; 以下详细描述了各个步骤&#xff1a; 第一步&#xff1a;设置开发环境 确保你拥有以下工具&#xff1a; 文本编辑器&#xff1a;如Visual Studio Cod…

2278. 企鹅游行(最大流,拆点)

活动 - AcWing 在南极附近的某个地方&#xff0c;一些企鹅正站在一些浮冰上。 作为群居动物&#xff0c;企鹅们喜欢聚在一起&#xff0c;因此&#xff0c;它们想在同一块浮冰上会合。 企鹅们不想淋湿自己&#xff0c;所以它们只能利用自己有限的跳跃能力&#xff0c;在一块块…

容器_Docker ( 06 )

容器_Docker ( 05 ) Kubernetes 资源对象管理 资源对象文件 模板与帮助信息 资源对象文件优势 命令无法实现高级复杂的功能某些资源对象使用命令无法创建方便管理 , 保存 , 追溯历史 资源对象文件太长 , 记不住怎么办 使用命令创建模板查询帮助信息查询官方手册 生成资源…

区块链游戏解说:什么是 Ultimate Champions

作者&#xff1a;lesleyfootprint.network 编译&#xff1a;cicifootprint.network 数据源&#xff1a;Ultimate Champions Dashboard 什么是 Ultimate Champions Ultimate Champions 是一款免费的奇幻足球和篮球游戏&#xff0c;拥有官方授权的数字卡牌作为区块链上的 NFT…

go interface{} 和string的转换问题

1.遇到的问题 问题来源于,我sql模版拼接遇到的问题。 首先&#xff0c;这样是没有问题的。 var qhx interface{} "qhx"s : qhx.(string)fmt.Println(s) 但是当我在这段代码里用:1.类型断言 var sqlStr "select * from tx_user where username %s" join…

SpringBoot -【SmartInitializingSingleton】基础使用及应用场景

SmartInitializingSingleton 在继续深入探讨 SmartInitializingSingleton接口之前&#xff0c;让我们先了解一下 Spring Framework 的基本概念和背景。Spring Framework 是一个开源的 JavaEE&#xff08;Java Enterprise Edition&#xff09;全栈&#xff08;full-stack&#x…

C++面试题精选与解析

C面试题精选与解析 一、基础与语法 请问C中的指针和引用有什么区别&#xff1f; 指针是一个变量&#xff0c;存储的是另一个变量的内存地址。指针可以被重新赋值以指向另一个不同的对象。而引用是某个变量的别名&#xff0c;一旦引用被初始化为一个变量&#xff0c;就不能改变…

高级统计方法 第4次作业

作业评阅&#xff1a; 概念 2.问题 KNN分类和KNN回归都是KNN算法在不同类型数据上的应用&#xff0c;但它们之间存在明显的区别。 解决的问题类型不同&#xff1a;KNN分类适用于解决分类问题&#xff0c;而KNN回归则适用于解决回归问题。当响应变量是连续的&#xff0c;根据…