Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/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))
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement