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).
npx @tessl/cli install tessl/pypi-mysql-connector-python@9.4.0A 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.
pip install mysql-connector-pythonimport mysql.connectorFor async operations:
import mysql.connector.aioimport 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()MySQL Connector/Python is organized around several key components:
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: passKey connection parameters:
host: MySQL server host (default: 'localhost')port: MySQL server port (default: 3306)user: Username for authenticationpassword: Password for authenticationdatabase: Database name to connect tocharset: 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)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."""
passAvailable cursor types:
MySQLCursor: Standard tuple-based resultsMySQLCursorBuffered: Buffered results for immediate fetchingMySQLCursorDict: Dictionary-based results with column names as keysMySQLCursorRaw: Raw results without automatic type conversionMySQLCursorPrepared: Prepared statements for repeated executionManage 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."""
passPool configuration parameters:
pool_name: Unique identifier for the connection poolpool_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 connectionPerform 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]: passComprehensive 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."""
passHandle 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]): passMySQL-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."""
passSupport 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
}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