asmysql

• PyPI: https://pypi.org/project/asmysql/
• GitHub: https://github.com/vastxiao/asmysql
• Gitee: https://gitee.com/vastxiao/asmysql
• Docs: https://vastxiao.github.io/asmysql/

Introduction

AsMysql - Python asyncio async MySQL client/connection pool, built on top of aiomysql. Supports async queries, transactions, connection pool management, high performance, easy to use, production-ready.

Features

• Code supports type annotations.
• Very easy to use, simply inherit the AsMysql class for logical development.
• Supports automatic management of the MySQL connection pool and reconnection mechanism.
• Automatically captures and handles MysqlError errors globally.
• Separates MySQL connection engine and development logic class.
• Separates statement execution from data retrieval.
• Supports uncached data stream acquisition for large data result sets (without occupying memory).

Install

# Install from PyPI
pip install asmysql

Documentation

Quick Start v2

Using Engine class for MySQL connection:

import asyncio

from asmysql import Engine

# Create MySQL connection engine
engine = Engine(url="mysql://root:pass@127.0.0.1:3306/?charset=utf8mb4")


async def main():
    # Connect to MySQL
    await engine.connect()
    # Execute SQL statement
    async with engine.execute("select user,host from mysql.user") as result:
        async for item in result.iterate():
            print(item)
    # Disconnect MySQL connection
    await engine.disconnect()


asyncio.run(main())

Using AsMysql class for logical development:

import asyncio
from asmysql import Engine
from asmysql import AsMysql

# Write logical development class
class TestAsMysql(AsMysql):
    async def print_users(self):
        result = await self.client.execute('select user,host from mysql.user')
        if result.error:
            print(f"error_no: {result.error_no}, error_msg:{result.error_msg}")
        else:
            # result.iterate() is an asynchronous iterator that can fetch each row of the execution result.
            async for item in result.iterate():
                print(item)

async def main():
    # Create MySQL connection engine
    engine = Engine(host='192.168.1.192', port=3306)
    # Connect to MySQL
    await engine.connect()
    # Create logical development class instance
    test_mysql = TestAsMysql(engine)
    # Execute logic
    await test_mysql.print_users()
    # Disconnect MySQL connection
    await engine.disconnect()

asyncio.run(main())

Compatible with v1 version

import asyncio
from asmysql.v1 import AsMysql

# Directly inherit the AsMysql class for development:
class TestAsMysql(AsMysql):
    # You can define some default parameters for the Mysql instance initialization here
    # The attributes are consistent with the __init__ parameters
    host = '127.0.0.1'
    port = 3306
    user = 'root'
    password = 'pass'

    async def get_users(self):
        # The self.client attribute is specifically used to execute SQL statements, providing aiomysql's execute and execute_many methods.
        result = await self.client.execute('select user,host from mysql.user')
        # result is specifically used to obtain execution results, providing fetch_one, fetch_many, fetch_all, and iterate methods.
        # result.err is the exception object (Exception) for all MySQL execution errors.
        if result.err:
            print(result.err_msg)
        else:
            # result.iterate() is an asynchronous iterator that can fetch each row of the execution result.
            async for item in result.iterate():
                print(item)

async def main():
    # This will create an instance and connect to MySQL:
    mysql = await TestAsMysql()
    # Execute SQL statement
    await mysql.get_users()
    # Remember to disconnect the MySQL connection before exiting the program:
    await mysql.disconnect()

asyncio.run(main())