CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-mysql-connector

MySQL driver written in Python providing comprehensive database connectivity with both traditional SQL and modern document operations

Pending
Overview
Eval results
Files

sql-operations.mddocs/

SQL Operations and Cursors

Traditional SQL operations with multiple cursor types for different data access patterns, including dictionary cursors, buffered cursors, prepared statements, and raw data access.

Capabilities

Standard Cursor Operations

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()

Dictionary Cursor

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']}")

Named Tuple Cursor

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}")

Buffered Cursors

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"""

Raw Cursors

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"""

Prepared Statements

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()

Stored Procedures

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}")

Batch Operations

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
    )
    """

C Extension Cursors

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 CMySQLCursorBuffered

Types

CursorConfig = {
    '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

docs

connections.md

django-integration.md

document-operations.md

error-handling.md

fabric-support.md

index.md

sql-operations.md

tile.json