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

—

Pending

Overview

Eval results

Files

MySQL Connector/Python

Name: tessl/pypi-mysql-connector-python
Author: tessl

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

Install with Tessl CLI

npx tessl i tessl/pypi-mysql-connector-python

Workspace: tessl
Visibility: Public
Created: 6 months ago
Last updated: about 1 month ago
Describes: pkg:pypi/mysql-connector-python@9.4.x
Publish Source: CLI
Badge