tessl/pypi-cymysql

Python MySQL Driver using Cython for high-performance database connectivity with async support

—

Pending

Overview

Eval results

Files

Cursor Operations

Name: tessl/pypi-cymysql
Author: tessl

SQL execution interface providing methods for running queries, managing transactions, and retrieving results with support for prepared statements and parameterized queries.

Capabilities

Cursor Creation and Management

Cursors are created from connection objects and provide the primary interface for executing SQL statements and retrieving results.

class Cursor:
    def __init__(self, connection):
        """
        Create a cursor object. Do not instantiate directly - use Connection.cursor().
        
        Parameters:
        - connection: Database connection object
        """
        
    def close(self):
        """
        Close the cursor and free associated resources.
        """
    
    @property
    def rowcount(self):
        """
        Number of rows affected by the last execute() operation.
        
        Returns:
        int: Row count, -1 if no execute() performed
        """
    
    @property
    def description(self):
        """
        Sequence describing columns in the result set.
        
        Returns:
        tuple: Column descriptions with (name, type_code, display_size, 
               internal_size, precision, scale, null_ok)
        """
    
    @property  
    def arraysize(self):
        """
        Number of rows to fetch at a time with fetchmany().
        
        Returns:
        int: Array size (default: 1)
        """
    
    @arraysize.setter
    def arraysize(self, value):
        """Set array size for fetchmany()."""
    
    @property
    def lastrowid(self):
        """
        The row ID of the last INSERT operation.
        
        Returns:
        int: ID of the last inserted row, or None if no INSERT performed
        """

SQL Execution

Execute SQL statements with optional parameter binding for safe query execution.

def execute(self, query, args=None):
    """
    Execute a SQL statement.
    
    Parameters:
    - query (str): SQL statement with optional %s placeholders
    - args (tuple/list/dict): Parameters to bind to placeholders
    
    Returns:
    int: Number of affected rows
    
    Raises:
    ProgrammingError: SQL syntax error
    OperationalError: Database operation error
    IntegrityError: Constraint violation
    DataError: Invalid data
    """

def executemany(self, query, args_list):
    """
    Execute a SQL statement multiple times with different parameter sets.
    
    Parameters:
    - query (str): SQL statement with %s placeholders
    - args_list (list/tuple): Sequence of parameter tuples/lists
    
    Returns:
    int: Number of affected rows from all executions
    
    Raises:
    ProgrammingError: SQL syntax error
    OperationalError: Database operation error
    """

def callproc(self, procname, args=()):
    """
    Call a stored procedure.
    
    Parameters:
    - procname (str): Stored procedure name
    - args (tuple): Procedure parameters
    
    Returns:
    tuple: Modified parameter list
    """

def nextset(self):
    """
    Move to the next result set when multiple result sets are available.
    
    Returns:
    bool: True if another result set is available, False otherwise
    """

Result Retrieval

Fetch query results in various formats and quantities.

def fetchone(self):
    """
    Fetch the next row from the result set.
    
    Returns:
    tuple: Next row data, or None if no more rows
    """

def fetchmany(self, size=None):
    """
    Fetch multiple rows from the result set.
    
    Parameters:
    - size (int): Number of rows to fetch (default: arraysize)
    
    Returns:
    list: List of row tuples, empty list if no more rows
    """

def fetchall(self):
    """
    Fetch all remaining rows from the result set.
    
    Returns:
    list: List of all remaining row tuples
    """

def scroll(self, value, mode='relative'):
    """
    Scroll the cursor position in the result set.
    
    Parameters:
    - value (int): Number of rows to move
    - mode (str): 'relative' or 'absolute' positioning mode
    """

def __iter__(self):
    """
    Iterator interface for row-by-row processing.
    
    Returns:
    Iterator: Cursor iterator yielding rows until None
    """

Context Manager Support

Cursors support context manager protocol for automatic cleanup:

def __enter__(self):
    """Enter context manager."""
    
def __exit__(self, exc_type, exc_val, exc_tb):
    """Exit context manager and close cursor."""

Dictionary Cursor

Dictionary cursor that returns results as dictionaries with column names as keys instead of tuples.

class DictCursor(Cursor):
    """
    Cursor that returns rows as dictionaries instead of tuples.
    
    Each row is returned as a dictionary mapping column names to values,
    making it easier to access specific columns by name.
    """
    
    def execute(self, query, args=None):
        """Execute query and prepare field mapping for dictionary results."""
        
    def fetchone(self):
        """
        Fetch the next row as a dictionary.
        
        Returns:
        dict: Row data as {column_name: value} mapping, or None if no more rows
        """
        
    def fetchmany(self, size=None):
        """
        Fetch multiple rows as dictionaries.
        
        Parameters:
        - size (int): Number of rows to fetch (default: arraysize)
        
        Returns:
        tuple: Tuple of dictionaries representing rows
        """
        
    def fetchall(self):
        """
        Fetch all remaining rows as dictionaries.
        
        Returns:
        tuple: Tuple of dictionaries representing all remaining rows
        """

Usage Examples

Basic Query Execution

import cymysql

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

# Execute query
cursor.execute("SELECT id, name FROM users WHERE age > %s", (18,))

# Fetch results
print(f"Query returned {cursor.rowcount} rows")
for row in cursor.fetchall():
    print(f"ID: {row[0]}, Name: {row[1]}")

cursor.close()
conn.close()

Parameterized Queries

# Safe parameter binding
cursor.execute(
    "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
    ('John Doe', 'john@example.com', 25)
)

# Dictionary parameter binding  
cursor.execute(
    "SELECT * FROM users WHERE name = %(name)s AND age > %(min_age)s",
    {'name': 'John', 'min_age': 18}
)

Batch Operations

# Insert multiple records efficiently
users_data = [
    ('Alice', 'alice@example.com', 30),
    ('Bob', 'bob@example.com', 25), 
    ('Charlie', 'charlie@example.com', 35)
]

cursor.executemany(
    "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)",
    users_data
)
print(f"Inserted {cursor.rowcount} records")

Result Set Navigation

cursor.execute("SELECT * FROM large_table")

# Process results in chunks
while True:
    rows = cursor.fetchmany(100)  # Fetch 100 rows at a time
    if not rows:
        break
        
    for row in rows:
        process_row(row)

Iterator Interface

cursor.execute("SELECT * FROM users")

# Use cursor as iterator
for row in cursor:
    print(f"Processing user: {row[1]}")

Context Manager Usage

conn = cymysql.connect(host='localhost', user='root', db='test')

with conn.cursor() as cursor:
    cursor.execute("SELECT COUNT(*) FROM users")
    count = cursor.fetchone()[0]
    print(f"Total users: {count}")
# Cursor automatically closed

Transaction Management

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

try:
    # Start transaction
    cursor.execute("START TRANSACTION")
    
    # Execute multiple statements
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = %s", (1,))
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = %s", (2,))
    
    # Commit if all successful
    conn.commit()
    print("Transaction committed")
    
except Exception as e:
    # Rollback on error
    conn.rollback()
    print(f"Transaction rolled back: {e}")
finally:
    cursor.close()
    conn.close()

Stored Procedure Calls

cursor = conn.cursor()

# Call stored procedure
result_args = cursor.callproc('GetUserStats', (user_id, 2023))
print(f"Procedure returned: {result_args}")

# Fetch procedure results
for row in cursor.fetchall():
    print(row)

Column Information

cursor.execute("SELECT id, name, email FROM users LIMIT 1")

# Access column metadata
print("Column information:")
for desc in cursor.description:
    print(f"  {desc[0]}: {desc[1]} (size: {desc[3]})")

Dictionary Cursor Usage

import cymysql
from cymysql.cursors import DictCursor

conn = cymysql.connect(host='localhost', user='root', db='test')

# Create dictionary cursor
cursor = conn.cursor(DictCursor)

# Execute query - results returned as dictionaries
cursor.execute("SELECT id, name, email FROM users WHERE active = %s", (True,))

# Fetch results as dictionaries
for row in cursor.fetchall():
    print(f"User {row['name']} has email {row['email']} (ID: {row['id']})")

# Access specific columns by name
cursor.execute("SELECT COUNT(*) as user_count FROM users")
result = cursor.fetchone()
print(f"Total users: {result['user_count']}")

cursor.close()
conn.close()

Lastrowid Example

import cymysql

conn = cymysql.connect(host='localhost', user='root', db='test')
cursor = conn.cursor()

# Insert new record and get the ID
cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", 
               ('New User', 'new@example.com'))

# Get the ID of the inserted row
new_user_id = cursor.lastrowid
print(f"Inserted user with ID: {new_user_id}")

conn.commit()
cursor.close()
conn.close()