tessl/pypi-aiomysql
MySQL driver for asyncio providing async/await support for database operations.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
connections.mddocs/
Database Connections
Create and manage individual MySQL database connections with full async/await support. Connections provide the foundation for all database operations in aiomysql.
Capabilities
Connection Creation
Create a new MySQL database connection with comprehensive configuration options.
def connect(
host: str = "localhost",
user: str = None,
password: str = "",
db: str = None,
port: int = 3306,
unix_socket: str = None,
charset: str = '',
sql_mode: str = None,
read_default_file: str = None,
conv: dict = None,
use_unicode: bool = None,
client_flag: int = 0,
cursorclass: type = Cursor,
init_command: str = None,
connect_timeout: int = None,
read_default_group: str = None,
autocommit: bool = False,
echo: bool = False,
local_infile: bool = False,
loop = None,
ssl: dict = None,
auth_plugin: str = '',
program_name: str = '',
server_public_key: str = None
) -> _ConnectionContextManager:
"""
Create a MySQL database connection.
Parameters:
- host: MySQL server hostname
- user: Username for authentication
- password: Password for authentication
- db: Database name to connect to
- port: MySQL server port (default 3306)
- unix_socket: Unix socket path for local connections
- charset: Character set for the connection
- sql_mode: SQL mode for the session
- read_default_file: MySQL configuration file to read
- conv: Custom type converters dictionary
- use_unicode: Enable Unicode support
- client_flag: MySQL client flags
- cursorclass: Default cursor class for new cursors
- init_command: SQL command to run on connection
- connect_timeout: Connection timeout in seconds
- read_default_group: Configuration file group to read
- autocommit: Enable autocommit mode
- echo: Enable query logging
- local_infile: Enable LOAD DATA LOCAL INFILE
- loop: Event loop to use
- ssl: SSL configuration dictionary
- auth_plugin: Authentication plugin name
- program_name: Program name for connection
- server_public_key: Server's public key for authentication
Returns:
Connection object
"""Connection Management
The Connection class provides methods for managing database connections and executing transactions.
class Connection:
@property
def closed(self) -> bool: ...
@property
def host(self) -> str: ...
@property
def port(self) -> int: ...
@property
def user(self) -> str: ...
@property
def db(self) -> str: ...
@property
def charset(self) -> str: ...
@property
def encoding(self) -> str: ...
def cursor(self, cursor=None) -> Cursor:
"""
Create a new cursor for executing queries.
Parameters:
- cursor: Cursor class to use (default: connection's cursorclass)
Returns:
Cursor instance
"""
async def ping(self, reconnect: bool = True) -> None:
"""
Check if the server is alive and optionally reconnect.
Parameters:
- reconnect: Whether to reconnect if connection is lost
"""
def close(self) -> None:
"""
Close the connection immediately without sending quit command.
"""
async def ensure_closed(self) -> None:
"""
Send quit command and close the connection gracefully.
"""
async def autocommit(self, value: bool) -> None:
"""
Enable or disable autocommit mode.
Parameters:
- value: True to enable autocommit, False to disable
"""
async def begin(self) -> None:
"""
Begin a new transaction.
"""
async def commit(self) -> None:
"""
Commit the current transaction.
"""
async def rollback(self) -> None:
"""
Rollback the current transaction.
"""
async def select_db(self, db: str) -> None:
"""
Change the default database for the connection.
Parameters:
- db: Database name to switch to
"""
async def show_warnings(self) -> tuple:
"""
Retrieve MySQL warnings from the last executed statement.
Returns:
Tuple of warning information
"""
async def set_charset(self, charset: str) -> None:
"""
Set the character set for the connection.
Parameters:
- charset: Character set name
"""
def escape(self, obj) -> str:
"""
Escape a value for safe use in SQL queries.
Parameters:
- obj: Value to escape
Returns:
Escaped string representation
"""
def literal(self, obj) -> str:
"""
Alias for escape() method.
Parameters:
- obj: Value to escape
Returns:
Escaped string representation
"""
def escape_string(self, s: str) -> str:
"""
Escape a string value for SQL queries.
Parameters:
- s: String to escape
Returns:
Escaped string
"""Usage Examples
Basic Connection
import asyncio
import aiomysql
async def connect_example():
# Create connection
conn = await aiomysql.connect(
host='localhost',
port=3306,
user='myuser',
password='mypass',
db='mydatabase'
)
# Use connection
async with conn.cursor() as cur:
await cur.execute("SELECT VERSION()")
version = await cur.fetchone()
print(f"MySQL version: {version[0]}")
# Close connection
conn.close()
asyncio.run(connect_example())Connection with SSL
async def ssl_connection():
ssl_config = {
'ca': '/path/to/ca.pem',
'cert': '/path/to/client-cert.pem',
'key': '/path/to/client-key.pem'
}
conn = await aiomysql.connect(
host='secure-mysql-server.com',
user='myuser',
password='mypass',
db='mydatabase',
ssl=ssl_config
)
# Use secure connection
async with conn.cursor() as cur:
await cur.execute("SELECT * FROM sensitive_data LIMIT 1")
data = await cur.fetchone()
print(data)
conn.close()Transaction Management
async def transaction_example():
conn = await aiomysql.connect(
host='localhost',
user='myuser',
password='mypass',
db='mydatabase',
autocommit=False # Disable autocommit for explicit transactions
)
try:
# Begin transaction
await conn.begin()
async with conn.cursor() as cur:
# Execute multiple statements in transaction
await cur.execute("INSERT INTO users (name, email) VALUES (%s, %s)",
("John Doe", "john@example.com"))
await cur.execute("INSERT INTO profiles (user_id, bio) VALUES (LAST_INSERT_ID(), %s)",
("Software developer",))
# Commit transaction
await conn.commit()
print("Transaction committed successfully")
except Exception as e:
# Rollback on error
await conn.rollback()
print(f"Transaction rolled back: {e}")
finally:
conn.close()Install with Tessl CLI
npx tessl i tessl/pypi-aiomysql