CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-mysql-connector-python

A self-contained Python driver for communicating with MySQL servers, using an API that is compliant with the Python Database API Specification v2.0 (PEP 249).

—

Pending
Overview
Eval results
Files

cursors.mddocs/

Cursor Operations

Execute SQL statements and retrieve results using various cursor types optimized for different use cases and data access patterns.

Cursor Types Overview

MySQL Connector/Python provides several cursor types:

  • MySQLCursor: Standard cursor returning tuples
  • MySQLCursorBuffered: Buffered cursor for immediate result fetching
  • MySQLCursorDict: Dictionary cursor with column names as keys
  • MySQLCursorRaw: Raw cursor without automatic type conversion
  • MySQLCursorPrepared: Prepared statement cursor for repeated execution
  • C Extension Cursors: High-performance variants when C extension is available

Creating Cursors

# From connection object
cursor = connection.cursor(
    buffered: Optional[bool] = None,       # Buffer all results immediately
    raw: Optional[bool] = None,            # Return raw MySQL types
    prepared: Optional[bool] = None,       # Use prepared statements
    cursor_class: Optional[Type] = None,   # Custom cursor class
    dictionary: Optional[bool] = None,     # Return results as dictionaries
    named_tuple: Optional[bool] = None     # Return results as named tuples
)

Base Cursor Classes

MySQLCursor

class MySQLCursor:
    """
    Standard cursor for executing SQL statements.
    Returns results as tuples.
    """
    
    def execute(self, operation: str, params: Optional[Union[Sequence, Dict]] = None, multi: bool = False) -> Optional[Iterator]:
        """Execute SQL statement with optional parameters."""
        pass
    
    def executemany(self, operation: str, seq_params: Sequence[Union[Sequence, Dict]]) -> None:
        """Execute SQL statement multiple times with parameter sequences."""
        pass
    
    def fetchone(self) -> Optional[Tuple]:
        """Fetch next row from result set."""
        pass
    
    def fetchmany(self, size: Optional[int] = None) -> List[Tuple]:
        """Fetch specified number of rows from result set."""
        pass
    
    def fetchall(self) -> List[Tuple]:
        """Fetch all remaining rows from result set."""
        pass
    
    def close(self) -> None:
        """Close cursor and free resources."""
        pass
    
    def callproc(self, procname: str, args: Sequence = ()) -> Optional[Dict]:
        """Call stored procedure with arguments."""
        pass
    
    def stored_results(self) -> Iterator['MySQLCursor']:
        """Return iterator for stored procedure result sets."""
        pass
    
    def nextset(self) -> Optional[bool]:
        """Skip to next result set in multi-result query."""
        pass
    
    def setinputsizes(self, sizes: Sequence) -> None:
        """Set input parameter sizes (DB-API compliance)."""
        pass
    
    def setoutputsize(self, size: int, column: Optional[int] = None) -> None:
        """Set output column size (DB-API compliance)."""
        pass
    
    @property 
    def description(self) -> Optional[List[Tuple]]:
        """Column metadata for last executed query."""
        pass
    
    @property
    def rowcount(self) -> int:
        """Number of rows affected by last operation."""
        pass
    
    @property
    def lastrowid(self) -> Optional[int]:
        """Auto-generated ID from last INSERT operation.""" 
        pass
    
    @property
    def arraysize(self) -> int:
        """Default number of rows fetchmany() should return."""
        pass
    
    @arraysize.setter  
    def arraysize(self, value: int) -> None:
        """Set default fetchmany() size."""
        pass
    
    @property
    def statement(self) -> Optional[str]:
        """Last executed SQL statement."""
        pass
    
    @property
    def with_rows(self) -> bool:
        """Whether last operation produced result rows."""
        pass
    
    @property
    def column_names(self) -> Tuple[str, ...]:
        """Column names from result set."""
        pass
    
    def __iter__(self) -> Iterator[Tuple]:
        """Make cursor iterable over result rows."""
        pass
    
    def __next__(self) -> Tuple:
        """Get next row for iteration."""
        pass
    
    def __enter__(self) -> 'MySQLCursor':
        """Context manager entry."""
        pass
    
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """Context manager exit with automatic cleanup."""
        pass

MySQLCursorBuffered

class MySQLCursorBuffered(MySQLCursor):
    """
    Buffered cursor that fetches all results immediately.
    Useful when you need to access all rows or get accurate rowcount.
    """
    
    @property
    def rowcount(self) -> int:
        """Accurate row count (buffered results)."""
        pass
    
    def reset(self, free: bool = True) -> None:
        """Reset cursor position to beginning of result set."""
        pass

MySQLCursorRaw

class MySQLCursorRaw(MySQLCursor):
    """
    Raw cursor that returns results without automatic type conversion.
    MySQL values returned as received from server.
    """
    
    def fetchone(self) -> Optional[Tuple]:
        """Fetch next row as raw MySQL types."""
        pass
    
    def fetchmany(self, size: Optional[int] = None) -> List[Tuple]:
        """Fetch rows as raw MySQL types."""
        pass
    
    def fetchall(self) -> List[Tuple]:
        """Fetch all rows as raw MySQL types."""
        pass

MySQLCursorBufferedRaw

class MySQLCursorBufferedRaw(MySQLCursorBuffered, MySQLCursorRaw):
    """
    Combination of buffered and raw cursor features.
    Buffers all results as raw MySQL types.
    """
    pass

Dictionary Cursors

MySQLCursorDict

class MySQLCursorDict(MySQLCursor):
    """
    Dictionary cursor returning results as dictionaries.
    Column names used as dictionary keys.
    """
    
    def fetchone(self) -> Optional[Dict[str, Any]]:
        """Fetch next row as dictionary."""
        pass
    
    def fetchmany(self, size: Optional[int] = None) -> List[Dict[str, Any]]:
        """Fetch rows as list of dictionaries."""
        pass
    
    def fetchall(self) -> List[Dict[str, Any]]:
        """Fetch all rows as list of dictionaries."""
        pass

MySQLCursorBufferedDict

class MySQLCursorBufferedDict(MySQLCursorBuffered, MySQLCursorDict):
    """
    Buffered dictionary cursor.
    Combines buffering with dictionary result format.
    """
    pass

Prepared Statement Cursors

MySQLCursorPrepared

class MySQLCursorPrepared(MySQLCursor):
    """
    Prepared statement cursor for repeated execution.
    Offers better performance for repeated queries.
    """
    
    def prepare(self, operation: str) -> None:
        """Prepare SQL statement for execution."""
        pass
    
    def execute(self, params: Optional[Union[Sequence, Dict]] = None) -> None:
        """Execute prepared statement with parameters."""
        pass
    
    def executemany(self, seq_params: Sequence[Union[Sequence, Dict]]) -> None:
        """Execute prepared statement multiple times."""
        pass
    
    @property
    def statement(self) -> Optional[str]:
        """Prepared SQL statement."""
        pass

MySQLCursorPreparedDict

class MySQLCursorPreparedDict(MySQLCursorPrepared):
    """
    Prepared statement cursor returning dictionary results.
    """
    
    def fetchone(self) -> Optional[Dict[str, Any]]:
        """Fetch next row as dictionary."""
        pass
    
    def fetchmany(self, size: Optional[int] = None) -> List[Dict[str, Any]]:
        """Fetch rows as list of dictionaries."""
        pass
    
    def fetchall(self) -> List[Dict[str, Any]]:
        """Fetch all rows as list of dictionaries."""
        pass

C Extension Cursors

When the C extension is available (HAVE_CEXT = True), high-performance cursor variants are provided:

class CMySQLCursor(MySQLCursor):
    """C extension cursor for improved performance."""
    pass

class CMySQLCursorBuffered(MySQLCursorBuffered):
    """C extension buffered cursor."""
    pass

class CMySQLCursorRaw(MySQLCursorRaw):
    """C extension raw cursor."""
    pass

class CMySQLCursorBufferedRaw(MySQLCursorBufferedRaw):
    """C extension buffered raw cursor."""
    pass

class CMySQLCursorDict(MySQLCursorDict):
    """C extension dictionary cursor."""
    pass

class CMySQLCursorBufferedDict(MySQLCursorBufferedDict):
    """C extension buffered dictionary cursor."""
    pass

class CMySQLCursorPrepared(MySQLCursorPrepared):
    """C extension prepared statement cursor."""
    pass

Usage Examples

Basic Query Execution

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser',
    password='mypassword',
    database='mydatabase'
)

# Standard cursor
cursor = connection.cursor()
cursor.execute("SELECT id, name, email FROM users WHERE age > %s", (25,))

# Fetch results
for (user_id, name, email) in cursor:
    print(f"User {user_id}: {name} ({email})")

cursor.close()
connection.close()

Dictionary Cursor Usage

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser',
    password='mypassword',
    database='mydatabase'
)

# Dictionary cursor
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT id, name, email FROM users WHERE age > %s", (25,))

# Results as dictionaries
for row in cursor:
    print(f"User {row['id']}: {row['name']} ({row['email']})")

cursor.close()
connection.close()

Buffered Cursor for Row Count

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser',
    password='mypassword',
    database='mydatabase'
)

# Buffered cursor provides accurate row count
cursor = connection.cursor(buffered=True)
cursor.execute("SELECT * FROM users")

print(f"Total users: {cursor.rowcount}")

# Fetch results
users = cursor.fetchall()
for user in users:
    print(user)

cursor.close()
connection.close()

Prepared Statements

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser',
    password='mypassword',
    database='mydatabase'
)

# Prepared statement cursor
cursor = connection.cursor(prepared=True)

# Prepare statement once
add_user = "INSERT INTO users (name, email, age) VALUES (?, ?, ?)"

# Execute multiple times with different parameters
user_data = [
    ('John Doe', 'john@example.com', 30),
    ('Jane Smith', 'jane@example.com', 25),
    ('Bob Johnson', 'bob@example.com', 35)
]

for data in user_data:
    cursor.execute(add_user, data)

connection.commit()
cursor.close()
connection.close()

Multi-Statement Execution

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser',
    password='mypassword',
    database='mydatabase'
)

cursor = connection.cursor()

# Execute multiple statements
sql_statements = """
    CREATE TEMPORARY TABLE temp_users (id INT, name VARCHAR(100));
    INSERT INTO temp_users VALUES (1, 'Test User');
    SELECT * FROM temp_users;
"""

# multi=True returns iterator for multiple results
results = cursor.execute(sql_statements, multi=True)

for result in results:
    if result.with_rows:
        print(f"Rows produced: {result.fetchall()}")
    else:
        print(f"Rows affected: {result.rowcount}")

cursor.close()
connection.close()

Stored Procedure Calls

import mysql.connector

connection = mysql.connector.connect(
    host='localhost', 
    user='myuser',
    password='mypassword',
    database='mydatabase'
)

cursor = connection.cursor()

# Call stored procedure
result_args = cursor.callproc('get_user_stats', [2024])

# Get procedure results
for result in cursor.stored_results():
    stats = result.fetchall()
    print(f"User statistics: {stats}")

cursor.close()
connection.close()

Context Manager Usage

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser', 
    password='mypassword',
    database='mydatabase'
)

# Automatic cursor cleanup
with connection.cursor(dictionary=True) as cursor:
    cursor.execute("SELECT COUNT(*) as total FROM users")
    result = cursor.fetchone()
    print(f"Total users: {result['total']}")
    # Cursor automatically closed

connection.close()

Handling Large Result Sets

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser',
    password='mypassword', 
    database='mydatabase'
)

cursor = connection.cursor()
cursor.execute("SELECT * FROM large_table")

# Process results in chunks
while True:
    rows = cursor.fetchmany(1000)  # Fetch 1000 rows at a time
    if not rows:
        break
    
    for row in rows:
        # Process each row
        print(f"Processing row: {row[0]}")

cursor.close()
connection.close()

Raw Cursor for Performance

import mysql.connector

connection = mysql.connector.connect(
    host='localhost',
    user='myuser',
    password='mypassword',
    database='mydatabase'
)

# Raw cursor skips type conversion for better performance
cursor = connection.cursor(raw=True)
cursor.execute("SELECT id, created_at FROM logs")

for (raw_id, raw_timestamp) in cursor:
    # Values are returned as bytes/raw MySQL types
    user_id = int(raw_id)
    timestamp = raw_timestamp.decode('utf-8')
    print(f"Log entry {user_id} at {timestamp}")

cursor.close()
connection.close()

Install with Tessl CLI

npx tessl i tessl/pypi-mysql-connector-python

docs

async.md

auth.md

connection.md

constants.md

cursors.md

errors.md

index.md

pooling.md

types.md

utilities.md

tile.json