MySQL driver written in Python providing comprehensive database connectivity with both traditional SQL and modern document operations
—
Traditional SQL operations with multiple cursor types for different data access patterns, including dictionary cursors, buffered cursors, prepared statements, and raw data access.
Basic cursor functionality for executing SQL statements and retrieving results.
class MySQLCursor:
"""
Standard cursor for executing SQL statements.
Provides basic functionality for SQL operations including SELECT, INSERT,
UPDATE, DELETE statements with parameter binding and result fetching.
"""
def execute(self, statement, params=None, multi=False):
"""
Execute a SQL statement.
Parameters:
- statement (str): SQL statement to execute
- params (tuple/list/dict): Parameters for statement
- multi (bool): Execute multiple statements
Returns:
iterator: Iterator over results for multi-statement execution
Raises:
ProgrammingError: Invalid SQL statement
DatabaseError: Database execution error
"""
def executemany(self, statement, seq_params):
"""
Execute statement for each parameter sequence.
Parameters:
- statement (str): SQL statement with parameter placeholders
- seq_params (list): Sequence of parameter tuples/dicts
"""
def fetchone(self):
"""
Fetch next row from result set.
Returns:
tuple or None: Next row or None if no more rows
"""
def fetchmany(self, size=1):
"""
Fetch multiple rows from result set.
Parameters:
- size (int): Number of rows to fetch
Returns:
list: List of row tuples
"""
def fetchall(self):
"""
Fetch all remaining rows from result set.
Returns:
list: List of all remaining row tuples
"""
def close(self):
"""Close the cursor and free resources"""
def callproc(self, procname, args=()):
"""
Call a stored procedure.
Parameters:
- procname (str): Stored procedure name
- args (tuple): Procedure arguments
Returns:
tuple: Modified arguments from procedure
"""
def nextset(self):
"""
Move to next result set in multi-result query.
Returns:
bool: True if more result sets available
"""
def setinputsizes(self, sizes):
"""Set input parameter sizes (DB-API compatibility)"""
def setoutputsize(self, size, column=None):
"""Set output buffer size (DB-API compatibility)"""
# Properties
@property
def description(self):
"""Column descriptions for last executed query"""
@property
def rowcount(self):
"""Number of rows affected by last operation"""
@property
def lastrowid(self):
"""Row ID of last inserted row"""
@property
def statement(self):
"""Last executed SQL statement"""Usage Example:
import mysql.connector
connection = mysql.connector.connect(
user='myuser',
password='mypassword',
host='localhost',
database='mydatabase'
)
cursor = connection.cursor()
# Execute SELECT query
cursor.execute("SELECT id, name, email FROM users WHERE age > %s", (18,))
# Fetch results
for (user_id, name, email) in cursor:
print(f"User {user_id}: {name} ({email})")
# Execute INSERT with parameters
cursor.execute(
"INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
("John Doe", "john@example.com", 25)
)
connection.commit()
cursor.close()
connection.close()Returns query results as dictionaries with column names as keys.
class MySQLCursorDict(MySQLCursor):
"""
Dictionary cursor returning rows as dictionaries.
Each row is returned as a dictionary where keys are column names
and values are the corresponding field values.
"""
def fetchone(self):
"""
Fetch next row as dictionary.
Returns:
dict or None: Row as dictionary or None if no more rows
"""
def fetchmany(self, size=1):
"""
Fetch multiple rows as list of dictionaries.
Returns:
list[dict]: List of row dictionaries
"""
def fetchall(self):
"""
Fetch all rows as list of dictionaries.
Returns:
list[dict]: List of all row dictionaries
"""
class MySQLCursorBufferedDict(MySQLCursorDict):
"""Buffered dictionary cursor that fetches all results immediately"""Usage Example:
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT id, name, email FROM users LIMIT 5")
users = cursor.fetchall()
for user in users:
print(f"ID: {user['id']}, Name: {user['name']}, Email: {user['email']}")Returns query results as named tuples with column names as attributes.
class MySQLCursorNamedTuple(MySQLCursor):
"""
Named tuple cursor returning rows as named tuples.
Each row is returned as a named tuple where field names correspond
to column names, allowing both index and attribute access.
"""
def fetchone(self):
"""
Fetch next row as named tuple.
Returns:
namedtuple or None: Row as named tuple or None
"""
def fetchmany(self, size=1):
"""
Fetch multiple rows as list of named tuples.
Returns:
list[namedtuple]: List of row named tuples
"""
def fetchall(self):
"""
Fetch all rows as list of named tuples.
Returns:
list[namedtuple]: List of all row named tuples
"""
class MySQLCursorBufferedNamedTuple(MySQLCursorNamedTuple):
"""Buffered named tuple cursor that fetches all results immediately"""Usage Example:
cursor = connection.cursor(named_tuple=True)
cursor.execute("SELECT id, name, email FROM users LIMIT 5")
users = cursor.fetchall()
for user in users:
print(f"ID: {user.id}, Name: {user.name}, Email: {user.email}")Cursors that fetch and buffer all results immediately for improved performance.
class MySQLCursorBuffered(MySQLCursor):
"""
Buffered cursor that fetches all results immediately.
Improves performance by fetching all results at once and storing
them in memory. Useful for small to medium result sets.
"""
def with_rows(self):
"""Check if cursor has result rows"""
class MySQLCursorBufferedRaw(MySQLCursorBuffered):
"""Buffered cursor that returns raw data without type conversion"""Returns data exactly as received from MySQL server without type conversion.
class MySQLCursorRaw(MySQLCursor):
"""
Raw cursor returning data without type conversion.
Returns all values as received from MySQL server without Python
type conversion. Useful for performance-critical applications or
when custom type handling is required.
"""
class MySQLCursorBufferedRaw(MySQLCursorRaw):
"""Buffered raw cursor combining raw data with buffering"""Cursor supporting prepared statements for improved performance and security.
class MySQLCursorPrepared(MySQLCursor):
"""
Prepared statement cursor for improved performance and security.
Uses MySQL's prepared statement protocol for efficient execution
of repeated queries and automatic parameter escaping.
"""
def prepare(self, statement):
"""
Prepare a SQL statement for execution.
Parameters:
- statement (str): SQL statement with parameter markers (?)
"""
def execute(self, statement=None, params=None, multi=False):
"""
Execute prepared statement.
Parameters:
- statement (str): SQL statement or None to use prepared
- params (tuple/list): Parameters for statement
- multi (bool): Execute multiple statements
"""Usage Example:
cursor = connection.cursor(prepared=True)
# Prepare statement once
stmt = "SELECT id, name FROM users WHERE age > ? AND city = ?"
# Execute multiple times with different parameters
cursor.execute(stmt, (18, 'New York'))
results1 = cursor.fetchall()
cursor.execute(stmt, (25, 'Los Angeles'))
results2 = cursor.fetchall()Execute MySQL stored procedures and handle multiple result sets.
def callproc(self, procname, args=()):
"""
Execute a stored procedure.
Parameters:
- procname (str): Name of stored procedure to call
- args (tuple): Input parameters for procedure
Returns:
tuple: Modified arguments (including OUT parameters)
Example:
cursor.callproc('GetUserStats', (user_id, 0, 0))
# Returns tuple with IN/OUT parameter values
"""
def stored_results(self):
"""
Get iterator over result sets from stored procedure.
Returns:
iterator: Iterator over MySQLCursor objects for each result set
"""Usage Example:
cursor = connection.cursor()
# Call stored procedure
result_args = cursor.callproc('GetUserStats', (123, 0, 0))
print(f"Total users: {result_args[1]}, Active users: {result_args[2]}")
# Process multiple result sets
for result in cursor.stored_results():
rows = result.fetchall()
print(f"Result set: {rows}")Efficient execution of multiple similar operations.
def executemany(self, statement, seq_params):
"""
Execute statement multiple times with different parameters.
Parameters:
- statement (str): SQL statement with parameter placeholders
- seq_params (sequence): Sequence of parameter tuples/dicts
Example:
data = [
('John', 'john@example.com', 25),
('Jane', 'jane@example.com', 30),
('Bob', 'bob@example.com', 35)
]
cursor.executemany(
"INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
data
)
"""High-performance cursor implementations using the optional C extension for improved performance.
class CMySQLCursor(MySQLCursor):
"""
C extension cursor providing improved performance over pure Python implementation.
Requires the C extension to be available. Provides the same interface as
MySQLCursor but with optimized execution and data handling.
"""
class CMySQLCursorDict(MySQLCursorDict):
"""C extension dictionary cursor with improved performance"""
class CMySQLCursorBuffered(MySQLCursorBuffered):
"""C extension buffered cursor with improved performance"""
class CMySQLCursorBufferedDict(MySQLCursorBufferedDict):
"""C extension buffered dictionary cursor with improved performance"""
class CMySQLCursorRaw(MySQLCursorRaw):
"""C extension raw cursor with improved performance"""
class CMySQLCursorBufferedRaw(MySQLCursorBufferedRaw):
"""C extension buffered raw cursor with improved performance"""
class CMySQLCursorNamedTuple(MySQLCursorNamedTuple):
"""C extension named tuple cursor with improved performance"""
class CMySQLCursorBufferedNamedTuple(MySQLCursorBufferedNamedTuple):
"""C extension buffered named tuple cursor with improved performance"""
class CMySQLCursorPrepared(MySQLCursorPrepared):
"""C extension prepared statement cursor with improved performance"""Usage Example:
import mysql.connector
# Connect using C extension (when available)
connection = mysql.connector.connect(
user='myuser',
password='mypassword',
host='localhost',
database='mydatabase',
use_pure=False # Enable C extension
)
# C extension cursors are used automatically when use_pure=False
cursor = connection.cursor() # Returns CMySQLCursor
dict_cursor = connection.cursor(dictionary=True) # Returns CMySQLCursorDict
buffered_cursor = connection.cursor(buffered=True) # Returns CMySQLCursorBufferedCursorConfig = {
'buffered': bool,
'raw': bool,
'prepared': bool,
'cursor_class': type,
'dictionary': bool,
'named_tuple': bool
}
ColumnDescription = tuple[
str, # name
int, # type_code
int, # display_size
int, # internal_size
int, # precision
int, # scale
bool # null_ok
]Install with Tessl CLI
npx tessl i tessl/pypi-mysql-connector