or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

async.mdauth.mdconnection.mdconstants.mdcursors.mderrors.mdindex.mdpooling.mdtypes.mdutilities.md
tile.json

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/mysql-connector-python@9.4.x

To install, run

npx @tessl/cli install tessl/pypi-mysql-connector-python@9.4.0

index.mddocs/

MySQL Connector/Python

A comprehensive Python database driver for MySQL servers that provides both synchronous and asynchronous connectivity options. It implements the Python Database API Specification v2.0 (PEP 249) and includes a high-performance C extension for improved speed, making it suitable for enterprise applications, web frameworks, data analysis tools, and any Python application requiring robust MySQL database connectivity.

Package Information

  • Package Name: mysql-connector-python
  • Language: Python
  • Installation: pip install mysql-connector-python
  • Requires: Python 3.9-3.13
  • License: GNU GPLv2 (with FOSS License Exception)

Core Imports

import mysql.connector

For async operations:

import mysql.connector.aio

Basic Usage

import mysql.connector

# Connect to MySQL database
connection = mysql.connector.connect(
    host='localhost',
    user='username',
    password='password',
    database='mydatabase'
)

# Create cursor and execute query
cursor = connection.cursor()
cursor.execute("SELECT * FROM users WHERE age > %s", (25,))
results = cursor.fetchall()

# Clean up
cursor.close()
connection.close()

Architecture

MySQL Connector/Python is organized around several key components:

  • Connection Management: Core connection classes with optional C extension optimization
  • Cursor System: Multiple cursor types for different data access patterns
  • Connection Pooling: Efficient connection pool management for high-performance applications
  • Authentication: Comprehensive plugin system supporting all MySQL authentication methods
  • Error Handling: Complete exception hierarchy matching MySQL server and client errors
  • Type Conversion: Robust bidirectional conversion between Python and MySQL data types

Core Connection Management

Establish and manage database connections with comprehensive configuration options.

from typing import Any, List, Optional, Sequence, Tuple, Union

# Main connection function
def connect(**kwargs) -> Union[MySQLConnection, PooledMySQLConnection]:
    """Create connection to MySQL server with optional pooling."""
    pass

# Connection classes
class MySQLConnection:
    """Main synchronous connection class implementing DB-API 2.0."""
    pass

class CMySQLConnection:
    """C Extension optimized connection class (when available)."""
    pass

# Query Attributes API
def query_attrs() -> List[Tuple[str, Any]]: pass
def query_attrs_append(value: Tuple[str, Any]) -> None: pass
def query_attrs_remove(name: str) -> Any: pass
def query_attrs_clear() -> None: pass

# Server Information API
def get_server_info() -> Optional[str]: pass
def get_server_version() -> Optional[Tuple[int, ...]]: pass

# Security and State API
def is_secure() -> bool: pass
def have_next_result() -> bool: pass
def set_client_flags(flags: Union[int, Sequence[int]]) -> int: pass

Key connection parameters:

  • host: MySQL server host (default: 'localhost')
  • port: MySQL server port (default: 3306)
  • user: Username for authentication
  • password: Password for authentication
  • database: Database name to connect to
  • charset: Connection character set (default: 'utf8mb4')
  • use_unicode: Enable Unicode support (default: True)
  • ssl_disabled: Disable SSL/TLS (default: False)
  • autocommit: Enable autocommit mode (default: False)

Connection Management

Cursor Operations

Execute SQL statements and retrieve results with various cursor types optimized for different use cases.

# Standard cursors
class MySQLCursor:
    """Standard cursor for executing SQL statements."""
    def execute(self, operation: str, params: Optional[Sequence] = None) -> None: pass
    def fetchone(self) -> Optional[Tuple]: pass
    def fetchall(self) -> List[Tuple]: pass
    def fetchmany(self, size: int = 1) -> List[Tuple]: pass

class MySQLCursorDict:
    """Dictionary cursor returning results as dictionaries."""
    def fetchone(self) -> Optional[Dict]: pass
    def fetchall(self) -> List[Dict]: pass

class MySQLCursorPrepared:
    """Prepared statement cursor for repeated execution."""
    pass

Available cursor types:

  • MySQLCursor: Standard tuple-based results
  • MySQLCursorBuffered: Buffered results for immediate fetching
  • MySQLCursorDict: Dictionary-based results with column names as keys
  • MySQLCursorRaw: Raw results without automatic type conversion
  • MySQLCursorPrepared: Prepared statements for repeated execution

Cursor Operations

Connection Pooling

Manage connection pools for high-performance applications with automatic connection lifecycle management.

class MySQLConnectionPool:
    """Connection pool manager with configurable size and behavior."""
    def __init__(self, pool_name: str, pool_size: int = 5, **kwargs): pass
    def get_connection(self) -> PooledMySQLConnection: pass
    def close(self) -> None: pass

class PooledMySQLConnection:
    """Pooled connection wrapper that returns to pool on close."""
    pass

def connect(**kwargs) -> PooledMySQLConnection:
    """Create pooled connection when pool parameters provided."""
    pass

Pool configuration parameters:

  • pool_name: Unique identifier for the connection pool
  • pool_size: Maximum number of connections in pool (default: 5)
  • pool_reset_session: Reset session variables on connection reuse (default: True)
  • pool_timeout: Maximum time to wait for available connection

Connection Pooling

Asynchronous Operations

Perform database operations asynchronously using asyncio with full async/await support.

import mysql.connector.aio

async def connect(**kwargs) -> MySQLConnection:
    """Create async connection to MySQL server."""
    pass

class MySQLConnection:
    """Async connection class with asyncio support."""
    async def connect(self) -> None: pass
    async def disconnect(self) -> None: pass
    def cursor(self, **kwargs) -> MySQLCursor: pass

class MySQLCursor:
    """Async cursor for executing SQL statements."""
    async def execute(self, operation: str, params: Optional[Sequence] = None) -> None: pass
    async def fetchone(self) -> Optional[Tuple]: pass
    async def fetchall(self) -> List[Tuple]: pass

Async Operations

Error Handling

Comprehensive exception hierarchy for robust error handling and debugging.

# Base exception classes
class Error(Exception):
    """Base exception class for all MySQL Connector errors."""
    pass

class DatabaseError(Error):
    """Database-related errors."""
    pass

class OperationalError(DatabaseError):
    """Operation-related errors (connection, queries)."""
    pass

class ProgrammingError(DatabaseError):
    """Programming errors (SQL syntax, parameter issues)."""
    pass

class IntegrityError(DatabaseError):
    """Database constraint violations."""
    pass

class DataError(DatabaseError):
    """Data processing errors."""
    pass

def custom_error_exception(**kwargs) -> Type[Error]:
    """Create custom error exceptions."""
    pass

Error Handling

Data Types and Conversion

Handle data type conversion between Python and MySQL with comprehensive type support.

# DB-API 2.0 type constructors
def Date(year: int, month: int, day: int) -> datetime.date: pass
def Time(hour: int, minute: int, second: int) -> datetime.time: pass
def Timestamp(year: int, month: int, day: int, hour: int, minute: int, second: int) -> datetime.datetime: pass
def Binary(string: bytes) -> bytes: pass

# Timestamp constructors from Unix time
def DateFromTicks(ticks: float) -> datetime.date: pass
def TimeFromTicks(ticks: float) -> datetime.time: pass
def TimestampFromTicks(ticks: float) -> datetime.datetime: pass

# Type objects for type comparison
STRING: Type
BINARY: Type
NUMBER: Type
DATETIME: Type
ROWID: Type

# Custom types
class HexLiteral:
    """Represents MySQL hexadecimal literals."""
    def __init__(self, value: Union[str, bytes]): pass

Data Types and Conversion

Constants and Enumerations

MySQL-specific constants for field types, flags, character sets, and configuration.

class FieldType:
    """MySQL field type constants."""
    DECIMAL: int
    TINY: int
    SHORT: int
    LONG: int
    FLOAT: int
    DOUBLE: int
    NULL: int
    TIMESTAMP: int
    LONGLONG: int
    INT24: int
    DATE: int
    TIME: int
    DATETIME: int
    YEAR: int
    VARCHAR: int
    BIT: int
    JSON: int
    NEWDECIMAL: int
    ENUM: int
    SET: int
    BLOB: int
    STRING: int
    GEOMETRY: int

class FieldFlag:
    """MySQL field flag constants."""
    NOT_NULL: int
    PRI_KEY: int
    UNIQUE_KEY: int
    MULTIPLE_KEY: int
    BLOB: int
    UNSIGNED: int
    ZEROFILL: int
    BINARY: int
    ENUM: int
    AUTO_INCREMENT: int
    TIMESTAMP: int
    SET: int

class ClientFlag:
    """MySQL client capability flags."""
    pass

class CharacterSet:
    """MySQL character set utilities."""
    pass

Constants and Enumerations

Authentication and Security

Support for all MySQL authentication methods including enterprise security features.

# Authentication plugins available:
# - mysql_native_password
# - caching_sha2_password  
# - sha256_password
# - mysql_clear_password
# - authentication_kerberos_client
# - authentication_ldap_sasl_client
# - authentication_oci_client
# - authentication_webauthn_client

# SSL/TLS configuration
ssl_config = {
    'ssl_cert': '/path/to/client-cert.pem',
    'ssl_key': '/path/to/client-key.pem', 
    'ssl_ca': '/path/to/ca-cert.pem',
    'ssl_verify_cert': True,
    'ssl_verify_identity': True
}

Authentication and Security

Utilities and Advanced Features

Low-level utilities, configuration management, and advanced functionality.

# Binary data utilities
def intread(buf: bytes) -> int: pass
def int1store(value: int) -> bytes: pass
def int2store(value: int) -> bytes: pass
def int4store(value: int) -> bytes: pass
def int8store(value: int) -> bytes: pass

# String utilities  
def read_lc_string(buf: bytes) -> Tuple[bytes, int]: pass
def read_string(buf: bytes, end: bytes, charset: str) -> str: pass

# Platform utilities
def get_platform() -> str: pass

# Configuration
class MySQLOptionsParser:
    """MySQL option file parser for my.cnf files."""
    def read_option_files(self, files: List[str]) -> Dict: pass

Utilities and Advanced Features