基于异步的快速操作MySQL的组件

These details have not been verified by PyPI

Project links

Project description

zapi_mysql

基于异步的快速操作MySQL的组件

项目地址： https://github.com/zhangdapeng520/zdpapi_mysql

使用pip安装

pip install zapi_mysql

一、增删改数据

1.1 创建表

import asyncio
from zapi_mysql import Mysql
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')


async def test_example_execute(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"
    
    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user
                                  (id INT,
                                  name VARCHAR(255),
                                  PRIMARY KEY (id));"""
    await db.execute(sql)
    
    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


loop = asyncio.get_event_loop()
loop.run_until_complete(test_example_execute(loop))

1.2 插入数据

import asyncio
from zapi_mysql import Mysql
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')


async def test_example_execute(loop):
    # 插入SQL语句
    sql = "INSERT INTO user VALUES(2,'李四')"
    await db.execute(sql)


loop = asyncio.get_event_loop()
loop.run_until_complete(test_example_execute(loop))

1.3 批量插入数据

import asyncio
from zapi_mysql import Mysql
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')


async def test_example_execute(loop):
    # 插入SQL语句
    data = [(4, 'gothic metal'), (5, 'doom metal'), (6, 'post metal')]
    sql = "INSERT INTO user VALUES(%s,%s)"
    await db.execute(sql, data=data)


loop = asyncio.get_event_loop()
loop.run_until_complete(test_example_execute(loop))

二、查询数据

2.1 查询所有数据

import asyncio
from zapi_mysql import Mysql
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')


async def test_example_execute(loop):
    # 插入SQL语句
    sql = "SELECT id, name FROM user ORDER BY id"
    result = await db.execute(sql)
    print("查询结果：\n", result)


loop = asyncio.get_event_loop()
loop.run_until_complete(test_example_execute(loop))

2.2 查询单条数据

import asyncio
from zapi_mysql import Mysql
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')


async def test_example_execute(loop):
    # 查询单条数据
    sql = "SELECT id, name FROM user ORDER BY id"
    result = await db.execute(sql, return_all=False)
    print("查询结果：\n", result)


loop = asyncio.get_event_loop()
loop.run_until_complete(test_example_execute(loop))

三、CRUD快捷工具

3.1 新增用户

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])

async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))

3.2 添加多条数据

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])

async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)
    

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))

3.3 根据ID删除数据

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])

async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)
    
async def test_delete(loop):
    # 根据ID删除数据
    await crud.delete(1)
    await crud.delete(2)
    await crud.delete(3)
    

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))
    loop.run_until_complete(test_delete(loop))

3.4 根据ID列表删除

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])

async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)
    
async def test_delete(loop):
    # 根据ID删除数据
    await crud.delete(1)
    await crud.delete(2)
    await crud.delete(3)
    
async def test_delete_ids(loop):
    # 根据ID列表删除数据
    await crud.delete_ids((3,4,5))
    

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))
    loop.run_until_complete(test_delete(loop))
    loop.run_until_complete(test_delete_ids(loop))

3.5 更新多条数据

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])

async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)
    
async def test_delete(loop):
    # 根据ID删除数据
    await crud.delete(1)
    await crud.delete(2)
    await crud.delete(3)
    
async def test_delete_ids(loop):
    # 根据ID列表删除数据
    await crud.delete_ids((3,4,5))
    
async def test_update(loop):
    # 根据ID更新数据
    await crud.update(6, {"name":"二郎神111"})
    
async def test_update_many(loop):
    # 更新多条数据
    data=[
        {"id":6, "name":"猪八戒"},
        {"id":7, "name":"嫦娥"},
    ]
    await crud.update_many(data)
   

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))
    loop.run_until_complete(test_delete(loop))
    loop.run_until_complete(test_delete_ids(loop))
    loop.run_until_complete(test_update(loop))
    loop.run_until_complete(test_update_many(loop))

3.6 查询单条数据

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])

async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)
    
async def test_delete(loop):
    # 根据ID删除数据
    await crud.delete(1)
    await crud.delete(2)
    await crud.delete(3)
    
async def test_delete_ids(loop):
    # 根据ID列表删除数据
    await crud.delete_ids((3,4,5))
    
async def test_update(loop):
    # 根据ID更新数据
    await crud.update(6, {"name":"二郎神111"})
    
async def test_update_many(loop):
    # 更新多条数据
    data=[
        {"id":6, "name":"猪八戒"},
        {"id":7, "name":"嫦娥"},
    ]
    await crud.update_many(data)

async def test_find_one(loop):
    # 查询单条数据
    result = await crud.find(6)
    print("查询结果：\n", result)
   

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))
    loop.run_until_complete(test_delete(loop))
    loop.run_until_complete(test_delete_ids(loop))
    loop.run_until_complete(test_update(loop))
    loop.run_until_complete(test_update_many(loop))
    loop.run_until_complete(test_find_one(loop))

3.8 根据ID列表查询

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])


async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)


async def test_delete(loop):
    # 根据ID删除数据
    await crud.delete(1)
    await crud.delete(2)
    await crud.delete(3)


async def test_delete_ids(loop):
    # 根据ID列表删除数据
    await crud.delete_ids((3, 4, 5))


async def test_update(loop):
    # 根据ID更新数据
    await crud.update(6, {"name": "二郎神111"})


async def test_update_many(loop):
    # 更新多条数据
    data = [
        {"id": 6, "name": "猪八戒"},
        {"id": 7, "name": "嫦娥"},
    ]
    await crud.update_many(data)


async def test_find_one(loop):
    # 查询单条数据
    result = await crud.find(6)
    print("查询结果：\n", result)


async def test_find_ids(loop):
    # 根据ID列表查询
    result = await crud.find_ids([6, 7, 8])
    print("查询结果：\n", result)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))
    loop.run_until_complete(test_delete(loop))
    loop.run_until_complete(test_delete_ids(loop))
    loop.run_until_complete(test_update(loop))
    loop.run_until_complete(test_update_many(loop))
    loop.run_until_complete(test_find_one(loop))
    loop.run_until_complete(test_find_ids(loop))

3.9 分页查询数据

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])


async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)


async def test_delete(loop):
    # 根据ID删除数据
    await crud.delete(1)
    await crud.delete(2)
    await crud.delete(3)


async def test_delete_ids(loop):
    # 根据ID列表删除数据
    await crud.delete_ids((3, 4, 5))


async def test_update(loop):
    # 根据ID更新数据
    await crud.update(6, {"name": "二郎神111"})


async def test_update_many(loop):
    # 更新多条数据
    data = [
        {"id": 6, "name": "猪八戒"},
        {"id": 7, "name": "嫦娥"},
    ]
    await crud.update_many(data)


async def test_find_one(loop):
    # 查询单条数据
    result = await crud.find(6)
    print("查询结果：\n", result)


async def test_find_ids(loop):
    # 根据ID列表查询
    result = await crud.find_ids([6, 7, 8])
    print("查询结果：\n", result)

async def test_find_page(loop):
    # 分页查询数据
    result = await crud.find_page(1, 20)
    print("查询结果：\n", result)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))
    loop.run_until_complete(test_delete(loop))
    loop.run_until_complete(test_delete_ids(loop))
    loop.run_until_complete(test_update(loop))
    loop.run_until_complete(test_update_many(loop))
    loop.run_until_complete(test_find_one(loop))
    loop.run_until_complete(test_find_ids(loop))
    loop.run_until_complete(test_find_page(loop))

3.10 查询总数

import asyncio
from zdpapi_mysql import Mysql, Crud
db = Mysql(host='127.0.0.1',
           port=3306,
           user='root',
           password='root',
           db='test')

crud = Crud(db, "user", ["name"])


async def test_create_table(loop):
    # 删除表
    await db.connect()
    sql = "DROP TABLE IF EXISTS user;"

    # 创建表
    await db.execute(sql)
    sql = """CREATE TABLE user(id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255));"""
    await db.execute(sql)

    # 插入SQL语句
    sql = "INSERT INTO user VALUES(1,'张三')"
    await db.execute(sql)


async def test_insert(loop):
    # 插入SQL语句
    await crud.add("李四")
    await crud.add("王五")
    await crud.add("赵六")


async def test_insert_many(loop):
    # 插入SQL语句
    data = [("孙悟空",), ("猪八戒",), ("沙僧",), ]
    await crud.add_many(data)


async def test_delete(loop):
    # 根据ID删除数据
    await crud.delete(1)
    await crud.delete(2)
    await crud.delete(3)


async def test_delete_ids(loop):
    # 根据ID列表删除数据
    await crud.delete_ids((3, 4, 5))


async def test_update(loop):
    # 根据ID更新数据
    await crud.update(6, {"name": "二郎神111"})


async def test_update_many(loop):
    # 更新多条数据
    data = [
        {"id": 6, "name": "猪八戒"},
        {"id": 7, "name": "嫦娥"},
    ]
    await crud.update_many(data)


async def test_find_one(loop):
    # 查询单条数据
    result = await crud.find(6)
    print("查询结果：\n", result)


async def test_find_ids(loop):
    # 根据ID列表查询
    result = await crud.find_ids([6, 7, 8])
    print("查询结果：\n", result)

async def test_find_page(loop):
    # 分页查询数据
    result = await crud.find_page(1, 20)
    print("查询结果：\n", result)

async def test_find_total(loop):
    # 查询数据总数
    result = await crud.find_total()
    print("查询结果：\n", result)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test_create_table(loop))
    loop.run_until_complete(test_insert(loop))
    loop.run_until_complete(test_insert_many(loop))
    loop.run_until_complete(test_delete(loop))
    loop.run_until_complete(test_delete_ids(loop))
    loop.run_until_complete(test_update(loop))
    loop.run_until_complete(test_update_many(loop))
    loop.run_until_complete(test_find_one(loop))
    loop.run_until_complete(test_find_ids(loop))
    loop.run_until_complete(test_find_page(loop))
    loop.run_until_complete(test_find_total(loop))

Project details

These details have not been verified by PyPI

Project links

Release history Release notifications | RSS feed

This version

1.0.2

Dec 5, 2021

1.0.1

Dec 4, 2021

1.0.0

Dec 4, 2021

0.1.0

Dec 3, 2021

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

zdpapi_mysql-1.0.2.tar.gz (66.3 kB view details)

Uploaded Dec 5, 2021 Source

Built Distribution

zdpapi_mysql-1.0.2-py3-none-any.whl (71.2 kB view details)

Uploaded Dec 5, 2021 Python 3

File details

Details for the file zdpapi_mysql-1.0.2.tar.gz.

File metadata

Download URL: zdpapi_mysql-1.0.2.tar.gz
Upload date: Dec 5, 2021
Size: 66.3 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: poetry/1.1.7 CPython/3.8.8 Windows/10

File hashes

Hashes for zdpapi_mysql-1.0.2.tar.gz
Algorithm	Hash digest
SHA256	`8a5204a32ee6774cfee1519c622cc5ff0c271a7a58e2044ae8eb0ce477cce65e`
MD5	`0990d1bc7e7101f936e98db06f9b2609`
BLAKE2b-256	`d3484e4f364e2be4454c382d6e0543d9002ecb613b19c4890635992da3bc4794`

See more details on using hashes here.

File details

Details for the file zdpapi_mysql-1.0.2-py3-none-any.whl.

File metadata

Download URL: zdpapi_mysql-1.0.2-py3-none-any.whl
Upload date: Dec 5, 2021
Size: 71.2 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: poetry/1.1.7 CPython/3.8.8 Windows/10

File hashes

Hashes for zdpapi_mysql-1.0.2-py3-none-any.whl
Algorithm	Hash digest
SHA256	`74203c9c64e8aa27c164c129b6485605a991160c265fc4971cfd0517a1d91c79`
MD5	`7751a8274438f08872afe4a17daa68f9`
BLAKE2b-256	`1602d5e1079a8bc2f9ca348cd3bfd4af0b64a91810e16b9c959dba023944ff02`