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).
—
Handle data type conversion between Python and MySQL with comprehensive type support, custom type definitions, and DB-API 2.0 compliance.
# Type aliases for standard datetime objects
Date = datetime.date
"""Alias for datetime.date class. Use as constructor: Date(year, month, day)"""
Time = datetime.time
"""Alias for datetime.time class. Use as constructor: Time(hour, minute, second, microsecond=0)"""
Timestamp = datetime.datetime
"""Alias for datetime.datetime class. Use as constructor: Timestamp(year, month, day, hour=0, minute=0, second=0, microsecond=0)"""def DateFromTicks(ticks: int) -> datetime.date:
"""
Create date from Unix timestamp.
Args:
ticks: Unix timestamp (seconds since epoch)
Returns:
datetime.date object
"""
pass
def TimeFromTicks(ticks: int) -> datetime.time:
"""
Create time from Unix timestamp.
Args:
ticks: Unix timestamp (seconds since epoch)
Returns:
datetime.time object (time portion only)
"""
pass
def TimestampFromTicks(ticks: int) -> datetime.datetime:
"""
Create datetime from Unix timestamp.
Args:
ticks: Unix timestamp (seconds since epoch)
Returns:
datetime.datetime object
"""
passBinary = bytes
"""Alias for bytes class. Use as constructor: Binary(data)"""Type objects for comparing column types in cursor descriptions:
# Type constants for column type comparison
STRING: Type = object() # String types (VARCHAR, CHAR, TEXT, etc.)
BINARY: Type = object() # Binary types (BLOB, BINARY, VARBINARY, etc.)
NUMBER: Type = object() # Numeric types (INT, DECIMAL, FLOAT, etc.)
DATETIME: Type = object() # Date/time types (DATE, TIME, DATETIME, TIMESTAMP)
ROWID: Type = object() # Row identifier typesclass HexLiteral:
"""
Represents MySQL hexadecimal literals for binary data.
Used for inserting binary data as hex strings in SQL.
"""
def __init__(self, value: Union[str, bytes]) -> None:
"""
Initialize hex literal.
Args:
value: String or bytes to represent as hex literal
"""
pass
@property
def value(self) -> bytes:
"""Get binary value."""
pass
def __str__(self) -> str:
"""String representation as hex literal."""
pass
def __repr__(self) -> str:
"""Developer representation."""
passclass MySQLConverterBase:
"""
Base class for data converters with type mapping functionality.
Defines interface for converting between Python and MySQL types.
"""
def __init__(self, charset: str = 'utf8mb4', use_unicode: bool = True) -> None:
"""Initialize converter with character set configuration."""
pass
def escape(self, value: Any) -> Union[str, bytes]:
"""
Escape value for safe inclusion in SQL statements.
Args:
value: Python value to escape
Returns:
Escaped string or bytes suitable for SQL
"""
pass
def quote(self, value: Any) -> str:
"""
Quote value for SQL statements.
Args:
value: Python value to quote
Returns:
Quoted string for SQL inclusion
"""
passclass MySQLConverter(MySQLConverterBase):
"""
Main data type converter between Python and MySQL types.
Handles bidirectional conversion with proper type mapping.
"""
def __init__(self, charset: str = 'utf8mb4', use_unicode: bool = True) -> None:
"""Initialize converter with character set and Unicode settings."""
pass
def to_mysql(self, value: Any) -> Union[str, bytes, None]:
"""
Convert Python value to MySQL format.
Args:
value: Python value to convert
Returns:
MySQL-compatible value
"""
pass
def from_mysql(self, value: Any, type_name: str) -> Any:
"""
Convert MySQL value to Python type.
Args:
value: MySQL value to convert
type_name: MySQL column type name
Returns:
Python value with appropriate type
"""
pass
# Specific type converters
def _str_to_mysql(self, value: str) -> bytes:
"""Convert string to MySQL bytes."""
pass
def _bytes_to_mysql(self, value: bytes) -> bytes:
"""Convert bytes to MySQL format."""
pass
def _int_to_mysql(self, value: int) -> str:
"""Convert integer to MySQL string."""
pass
def _float_to_mysql(self, value: float) -> str:
"""Convert float to MySQL string."""
pass
def _decimal_to_mysql(self, value: Decimal) -> str:
"""Convert Decimal to MySQL string."""
pass
def _bool_to_mysql(self, value: bool) -> str:
"""Convert boolean to MySQL string."""
pass
def _datetime_to_mysql(self, value: datetime.datetime) -> str:
"""Convert datetime to MySQL format."""
pass
def _date_to_mysql(self, value: datetime.date) -> str:
"""Convert date to MySQL format."""
pass
def _time_to_mysql(self, value: datetime.time) -> str:
"""Convert time to MySQL format."""
pass
def _timedelta_to_mysql(self, value: datetime.timedelta) -> str:
"""Convert timedelta to MySQL TIME format."""
pass
def _none_to_mysql(self, value: None) -> str:
"""Convert None to MySQL NULL."""
pass
# MySQL to Python converters
def _mysql_to_python_datetime(self, value: bytes, dsc: Any = None) -> datetime.datetime:
"""Convert MySQL DATETIME to Python datetime."""
pass
def _mysql_to_python_date(self, value: bytes, dsc: Any = None) -> datetime.date:
"""Convert MySQL DATE to Python date."""
pass
def _mysql_to_python_time(self, value: bytes, dsc: Any = None) -> datetime.time:
"""Convert MySQL TIME to Python time."""
pass
def _mysql_to_python_timestamp(self, value: bytes, dsc: Any = None) -> datetime.datetime:
"""Convert MySQL TIMESTAMP to Python datetime."""
pass
def _mysql_to_python_decimal(self, value: bytes, dsc: Any = None) -> Decimal:
"""Convert MySQL DECIMAL to Python Decimal."""
pass
def _mysql_to_python_int(self, value: bytes, dsc: Any = None) -> int:
"""Convert MySQL integer to Python int."""
pass
def _mysql_to_python_float(self, value: bytes, dsc: Any = None) -> float:
"""Convert MySQL float to Python float."""
pass
def _mysql_to_python_bit(self, value: bytes, dsc: Any = None) -> int:
"""Convert MySQL BIT to Python int."""
pass
def _mysql_to_python_json(self, value: bytes, dsc: Any = None) -> Union[Dict, List, str, int, float, bool, None]:
"""Convert MySQL JSON to Python object."""
pass# Type aliases for type hints
StrOrBytes = Union[str, bytes]
MySQLConvertibleType = Union[str, bytes, int, float, bool, datetime.datetime, datetime.date, datetime.time, Decimal, None]
DescriptionType = Tuple[str, int, Optional[int], Optional[int], Optional[int], Optional[int], bool]
RowType = Union[Tuple[Any, ...], Dict[str, Any]]
ParamsSequenceType = Union[Tuple[Any, ...], List[Any]]
ParamsDictType = Dict[str, Any]import mysql.connector
from mysql.connector import Date, Time, Timestamp, Binary
import datetime
# Using DB-API type constructors
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
# Insert with type constructors
insert_query = """
INSERT INTO events (name, event_date, event_time, created_at, binary_data)
VALUES (%s, %s, %s, %s, %s)
"""
data = (
'Conference',
Date(2024, 12, 25), # DATE column
Time(14, 30, 0), # TIME column
Timestamp(2024, 1, 15, 10, 30, 45), # DATETIME column
Binary(b'binary data content') # BLOB column
)
cursor.execute(insert_query, data)
connection.commit()
cursor.close()
connection.close()import mysql.connector
from mysql.connector import DateFromTicks, TimeFromTicks, TimestampFromTicks
import time
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
# Current Unix timestamp
current_timestamp = time.time()
# Convert to DB-API types
current_date = DateFromTicks(current_timestamp)
current_time = TimeFromTicks(current_timestamp)
current_datetime = TimestampFromTicks(current_timestamp)
cursor.execute("""
INSERT INTO log_entries (log_date, log_time, created_at)
VALUES (%s, %s, %s)
""", (current_date, current_time, current_datetime))
connection.commit()
cursor.close()
connection.close()import mysql.connector
from mysql.connector.custom_types import HexLiteral
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
# Binary data as hex literal
binary_data = b'\x00\x01\x02\x03\xFF'
hex_literal = HexLiteral(binary_data)
cursor.execute("INSERT INTO binary_table (data) VALUES (%s)", (hex_literal,))
connection.commit()
cursor.close()
connection.close()import mysql.connector
from mysql.connector import STRING, BINARY, NUMBER, DATETIME, ROWID
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
cursor.execute("SELECT id, name, email, created_at, profile_pic FROM users LIMIT 1")
# Check column types using description
for i, column in enumerate(cursor.description):
column_name = column[0]
column_type = column[1]
if column_type in NUMBER:
print(f"{column_name}: Numeric column")
elif column_type in STRING:
print(f"{column_name}: String column")
elif column_type in DATETIME:
print(f"{column_name}: Date/time column")
elif column_type in BINARY:
print(f"{column_name}: Binary column")
elif column_type in ROWID:
print(f"{column_name}: Row ID column")
else:
print(f"{column_name}: Other type")
cursor.close()
connection.close()import mysql.connector
from mysql.connector.conversion import MySQLConverter
import datetime
from decimal import Decimal
# Create custom converter
converter = MySQLConverter(charset='utf8mb4', use_unicode=True)
# Python to MySQL conversion examples
python_values = [
'Hello World',
42,
3.14159,
Decimal('999.99'),
True,
datetime.datetime(2024, 1, 15, 10, 30, 45),
datetime.date(2024, 1, 15),
datetime.time(10, 30, 45),
None,
b'binary data'
]
print("Python to MySQL conversions:")
for value in python_values:
mysql_value = converter.to_mysql(value)
print(f"{repr(value)} -> {repr(mysql_value)}")
# MySQL to Python conversion (conceptual - would use actual MySQL data)
print("\nMySQL to Python conversions:")
mysql_data = [
(b'2024-01-15 10:30:45', 'DATETIME'),
(b'2024-01-15', 'DATE'),
(b'10:30:45', 'TIME'),
(b'123.45', 'DECIMAL'),
(b'42', 'BIGINT'),
(b'3.14159', 'DOUBLE'),
(b'1', 'TINYINT'),
(b'Hello World', 'VARCHAR')
]
for mysql_value, type_name in mysql_data:
python_value = converter.from_mysql(mysql_value, type_name)
print(f"{repr(mysql_value)} ({type_name}) -> {repr(python_value)} ({type(python_value).__name__})")import mysql.connector
import json
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
# Insert JSON data
user_profile = {
'preferences': {
'theme': 'dark',
'language': 'en',
'notifications': True
},
'settings': {
'timezone': 'UTC',
'date_format': 'YYYY-MM-DD'
}
}
cursor.execute(
"INSERT INTO user_profiles (user_id, profile_data) VALUES (%s, %s)",
(123, json.dumps(user_profile))
)
# Retrieve and parse JSON data
cursor.execute("SELECT profile_data FROM user_profiles WHERE user_id = %s", (123,))
result = cursor.fetchone()
if result:
# MySQL Connector automatically converts JSON columns to Python objects
profile_data = result[0]
if isinstance(profile_data, str):
profile_data = json.loads(profile_data)
print(f"Theme: {profile_data['preferences']['theme']}")
print(f"Language: {profile_data['preferences']['language']}")
connection.commit()
cursor.close()
connection.close()import mysql.connector
from decimal import Decimal
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
# Use Decimal for precise financial calculations
price = Decimal('1999.99')
tax_rate = Decimal('0.08')
total = price * (1 + tax_rate)
cursor.execute("""
INSERT INTO orders (item_price, tax_rate, total_amount)
VALUES (%s, %s, %s)
""", (price, tax_rate, total))
# Retrieve Decimal values
cursor.execute("SELECT item_price, tax_rate, total_amount FROM orders ORDER BY id DESC LIMIT 1")
result = cursor.fetchone()
if result:
item_price, tax_rate, total_amount = result
print(f"Item Price: {item_price} (type: {type(item_price).__name__})")
print(f"Tax Rate: {tax_rate} (type: {type(tax_rate).__name__})")
print(f"Total: {total_amount} (type: {type(total_amount).__name__})")
connection.commit()
cursor.close()
connection.close()import mysql.connector
import datetime
import pytz
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase',
time_zone='+00:00' # Set connection timezone to UTC
)
cursor = connection.cursor()
# Create timezone-aware datetime
utc = pytz.UTC
eastern = pytz.timezone('US/Eastern')
# Current time in different timezones
now_utc = datetime.datetime.now(utc)
now_eastern = now_utc.astimezone(eastern)
# Store as UTC in database
cursor.execute("""
INSERT INTO events (name, event_time_utc, event_time_local)
VALUES (%s, %s, %s)
""", ('Meeting', now_utc.replace(tzinfo=None), now_eastern.replace(tzinfo=None)))
# Retrieve and handle timezone conversion
cursor.execute("SELECT event_time_utc FROM events ORDER BY id DESC LIMIT 1")
result = cursor.fetchone()
if result:
# Add timezone info back after retrieval
event_time_utc = result[0].replace(tzinfo=utc)
event_time_eastern = event_time_utc.astimezone(eastern)
print(f"Event time UTC: {event_time_utc}")
print(f"Event time Eastern: {event_time_eastern}")
connection.commit()
cursor.close()
connection.close()Install with Tessl CLI
npx tessl i tessl/pypi-mysql-connector-python