python aiomysql的例子

Posted 2021-05-09

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


import pprint
import asyncio
import aiomysql

loop = asyncio.get_event_loop()

async def test_example():
    conn = await aiomysql.connect(host='127.0.0.1', port=3306,
                                       user='test', password='test',
                                       db='test', loop=loop)
    cur = await conn.cursor()
    await cur.execute("SELECT * from service order by ser_id desc limit 20;")
    pprint.pprint(cur.description)
    r = await cur.fetchall()
    pprint.pprint(r)
    await cur.close()
    conn.close()

loop.run_until_complete(test_example())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


import asyncio
import warnings
import aiomysql
import pprint


async def stored_procedure(loop):
    conn = await aiomysql.connect(host='127.0.0.1', port=3306,
                                  user='test', password='test',
                                  db='test', loop=loop)
    async with conn.cursor() as cur:
        with warnings.catch_warnings():
            warnings.simplefilter('ignore')
            await cur.execute('DROP PROCEDURE IF EXISTS myinc;')
        await cur.execute("""CREATE PROCEDURE myinc(p1 INT)
                             BEGIN
                                SELECT p1 + 1;
                             END""")
        await cur.callproc('myinc', [1])
        (ret, ) = await cur.fetchone()
        assert 2, ret
        print(ret)

    conn.close()


async def test_example_executemany(loop):
    conn = await aiomysql.connect(host='127.0.0.1', port=3306,
                                  user='test', password='test',
                                  db='test', loop=loop)
    # cur = await conn.cursor()
    async with conn.cursor() as cur:
        await cur.execute("DROP TABLE IF EXISTS music_style;")
        await cur.execute("""CREATE TABLE music_style(
                                id INT,
                                name VARCHAR(255),
                                PRIMARY KEY(id));""")
        await conn.commit()

        await cur.execute("INSERT INTO music_style VALUES(1, 'heavy metal')")
        await cur.execute("INSERT INTO music_style VALUES(2, 'death metal')")
        await cur.execute("INSERT INTO music_style VALUES(3, 'power metal')")
        await conn.commit()

        data = [(4, 'gothic metal'), (5, 'doom metal'), (6, 'post metal')]
        await cur.executemany(
            "INSERT INTO music_style (id, name)"
            "VALUES (%s, %s)", data)
        await conn.commit()

        await cur.execute("SELECT * FROM music_style;")
        result = await cur.fetchall()
        print(result)


async def pool(loop):
    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306, user='test',
                                      password='test', db='test', loop=loop)
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT 42;")
            pprint.pprint(cur.description)
            (r, ) = await cur.fetchone()
            assert r == 42

    pool.close()
    await pool.wait_closed()


async def test_example_transaction(loop):
    conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='test',
                                  password='test', db='test', autocommit=False,
                                  loop=loop)
    async with conn.cursor() as cursor:
        stmt_drop = "DROP TABLE IF EXISTS names;"
        with warnings.catch_warnings():
            warnings.simplefilter('ignore')
            await cursor.execute(stmt_drop)
        await cursor.execute("""
            CREATE TABLE names (
                id TINYINT UNSIGNED NOT NULL AUTO_INCREMENT,
                name VARCHAR(30) DEFAULT '' NOT NULL,
                cnt TINYINT UNSIGNED DEFAULT 0,
                PRIMARY KEY(id)
            );""")
        await conn.commit()

        # Insert 3 records
        names = (('Geert', ), ('Jan', ), ('Michael', ))
        stmt_insert = "INSERT INTO names (name) VALUES (%s)"
        await cursor.executemany(stmt_insert, names)

        # Roll back!!!
        await conn.rollback()

        # There should be no data!
        stmt_select = "SELECT id, name FROM names ORDER BY id"
        await cursor.execute(stmt_select)
        resp = await cursor.fetchall()
        assert not resp

        await cursor.executemany(stmt_insert, names)

        await cursor.execute(stmt_select)
        await conn.commit()

        # Cannot rollback after the commit
        await conn.rollback()
        resp = await cursor.fetchall()
        assert resp
        print(resp)

        await cursor.execute(stmt_drop)
        await cursor.close()
        conn.close()


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