aiomysql

import asyncio import os import aiomysql async def main(): # Get credentials from environment variables for security db_host = os.environ.get('MYSQL_HOST', '127.0.0.1') db_user = os.environ.get('MYSQL_USER', 'root') db_password = os.environ.get('MYSQL_PASSWORD', 'password') db_name = os.environ.get('MYSQL_DB', 'test_db') try: # Establish an asynchronous connection async with await aiomysql.connect( host=db_host, user=db_user, password=db_password, db=db_name, autocommit=True # Or manage transactions manually ) as conn: print(f"Connected to MySQL on {db_host}") # Create a cursor object async with conn.cursor() as cursor: # Execute a query await cursor.execute("SELECT VERSION();") # Fetch one result version = await cursor.fetchone() print(f"MySQL Version: {version[0]}") # Execute another query (e.g., create a table) await cursor.execute( "CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255));" ) print("Table 'users' ensured to exist.") # Insert data await cursor.execute("INSERT INTO users (name) VALUES (%s);", ("Alice",)) await cursor.execute("INSERT INTO users (name) VALUES (%s);", ("Bob",)) print("Inserted Alice and Bob.") # Select data await cursor.execute("SELECT id, name FROM users;") users = await cursor.fetchall() print("Users:") for user_id, name in users: print(f" ID: {user_id}, Name: {name}") # The connection is automatically closed when exiting the 'async with' block except Exception as e: print(f"An error occurred: {e}") if __name__ == '__main__': asyncio.run(main())