tessl/pypi-cymysql

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

—

Pending

Overview

Eval results

Files

Database Connections

Name: tessl/pypi-cymysql
Author: tessl

Core connection functionality for establishing and managing database connections with support for SSL, authentication, character sets, and various MySQL connection parameters.

Capabilities

Connection Creation

Creates a new database connection with specified parameters. The connection object manages the underlying socket connection to the MySQL server and handles authentication, character set negotiation, and protocol initialization.

def connect(host="localhost", user=None, passwd="", db=None, port=3306, 
           unix_socket=None, charset='', sql_mode=None, read_default_file=None,
           client_flag=0, cursorclass=None, init_command=None, connect_timeout=None,
           ssl=None, read_default_group=None, compress="", zstd_compression_level=3,
           named_pipe=None, conv=None, encoders=None):
    """
    Create a database connection.
    
    Parameters:
    - host (str): MySQL server hostname or IP address (default: "localhost")
    - user (str): Username for authentication
    - passwd (str): Password for authentication (default: "")
    - port (int): MySQL server port number (default: 3306)
    - db (str): Default database name
    - unix_socket (str): Unix socket path for local connections
    - charset (str): Character set for connection (default: '')
    - sql_mode (str): SQL mode setting for connection
    - read_default_file (str): MySQL configuration file path
    - client_flag (int): Custom flags to send to MySQL
    - cursorclass: Default cursor class for this connection
    - init_command (str): SQL command to run on connection
    - connect_timeout (int): Connection timeout in seconds
    - ssl (dict): SSL configuration parameters
    - read_default_group (str): Group to read from configuration file
    - compress (str): Compression algorithm ("zlib" or "zstd")
    - zstd_compression_level (int): ZSTD compression level (1-22, default: 3)
    - named_pipe: Not supported (raises NotImplementedError)
    - conv: Decoders dictionary for custom type marshalling
    - encoders: Encoders dictionary for custom type marshalling
    
    Returns:
    Connection: Active database connection object
    
    Raises:
    OperationalError: Connection failed
    InterfaceError: Invalid connection parameters
    """

Connection Management

The Connection class provides methods for managing the database connection lifecycle, executing SQL statements, and controlling transaction behavior.

class Connection:
    def cursor(self, cursor=None):
        """
        Create a new cursor object for executing SQL statements.
        
        Parameters:
        - cursor: Cursor class to instantiate (optional)
        
        Returns:
        Cursor: New cursor object
        """
    
    def commit(self):
        """
        Commit current transaction.
        
        Raises:
        OperationalError: Transaction commit failed
        """
    
    def rollback(self):
        """
        Roll back current transaction.
        
        Raises:
        OperationalError: Transaction rollback failed
        """
    
    def close(self):
        """
        Close the database connection.
        """
    
    def autocommit(self, value):
        """
        Enable or disable autocommit mode.
        
        Parameters:
        - value (bool): True to enable autocommit, False to disable
        """
    
    def get_host_info(self):
        """
        Get information about the connection host.
        
        Returns:
        str: Host connection information
        """
    
    def get_proto_info(self):
        """
        Get MySQL protocol version.
        
        Returns:
        int: Protocol version number
        """
    
    def get_server_info(self):
        """
        Get MySQL server version information.
        
        Returns:
        str: Server version string
        """
    
    def open(self):
        """
        Check if connection is open.
        
        Returns:
        bool: True if connection is open, False otherwise
        """
    
    def ping(self, reconnect=True):
        """
        Check if connection to server is alive.
        
        Parameters:
        - reconnect (bool): Attempt to reconnect if connection is lost
        
        Returns:
        None
        
        Raises:
        OperationalError: Connection check failed
        """
    
    def set_charset(self, charset):
        """
        Set connection character set.
        
        Parameters:
        - charset (str): Character set name
        """
    
    def show_warnings(self):
        """
        Get warnings from last executed statement.
        
        Returns:
        tuple: Warning information
        """

Context Manager Support

Connections support Python context manager protocol for automatic resource cleanup:

# Automatic connection cleanup
with cymysql.connect(host='localhost', user='root', db='test') as conn:
    cur = conn.cursor()
    cur.execute("SELECT 1")
    result = cur.fetchone()
# Connection automatically closed when leaving context

SSL Configuration

CyMySQL supports SSL connections with various authentication modes:

# SSL connection with certificate verification
conn = cymysql.connect(
    host='mysql-server.example.com',
    user='secure_user',
    password='secure_password',
    ssl={
        'ca': '/path/to/ca.pem',
        'cert': '/path/to/client-cert.pem',
        'key': '/path/to/client-key.pem'
    },
    ssl_verify_cert=True,
    ssl_verify_identity=True
)

Connection Configuration Files

Support for MySQL configuration files for connection parameters:

# Load connection settings from my.cnf
conn = cymysql.connect(
    read_default_file='/etc/mysql/my.cnf',
    read_default_group='client'
)

Common Connection Patterns

Basic Database Connection

import cymysql

conn = cymysql.connect(
    host='localhost',
    user='myuser',
    password='mypassword', 
    database='mydatabase',
    charset='utf8mb4'
)

try:
    # Use connection
    cursor = conn.cursor()
    cursor.execute("SELECT VERSION()")
    version = cursor.fetchone()
    print(f"MySQL version: {version[0]}")
finally:
    conn.close()

Connection with Error Handling

import cymysql
from cymysql import OperationalError, ProgrammingError

try:
    conn = cymysql.connect(
        host='localhost',
        user='myuser',
        password='mypassword',
        database='mydatabase',
        connect_timeout=5
    )
    
    # Test connection
    conn.ping()
    print("Connection successful")
    
except OperationalError as e:
    print(f"Connection failed: {e}")
except ProgrammingError as e:
    print(f"Authentication error: {e}")

Install with Tessl CLI