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).
—
MySQL-specific constants for field types, flags, character sets, client capabilities, and configuration options that provide standardized access to MySQL protocol values.
class FieldType:
"""
MySQL field type constants for column type identification.
Used in cursor descriptions and type checking.
"""
# Numeric types
DECIMAL: int = 0 # DECIMAL or NUMERIC
TINY: int = 1 # TINYINT (-128 to 127 or 0 to 255)
SHORT: int = 2 # SMALLINT (-32768 to 32767 or 0 to 65535)
LONG: int = 3 # INTEGER (-2147483648 to 2147483647 or 0 to 4294967295)
FLOAT: int = 4 # FLOAT (single precision)
DOUBLE: int = 5 # DOUBLE (double precision)
LONGLONG: int = 8 # BIGINT (-9223372036854775808 to 9223372036854775807)
INT24: int = 9 # MEDIUMINT (-8388608 to 8388607 or 0 to 16777215)
NEWDECIMAL: int = 246 # Precision math DECIMAL or NUMERIC
BIT: int = 16 # BIT field type
# Date and time types
DATE: int = 10 # DATE (YYYY-MM-DD)
TIME: int = 11 # TIME (HH:MM:SS)
DATETIME: int = 12 # DATETIME (YYYY-MM-DD HH:MM:SS)
YEAR: int = 13 # YEAR (YYYY)
TIMESTAMP: int = 7 # TIMESTAMP (YYYY-MM-DD HH:MM:SS)
NEWDATE: int = 14 # Internal DATE type
# String and text types
VARCHAR: int = 15 # Variable-length string
STRING: int = 254 # Fixed-length string (CHAR)
VAR_STRING: int = 253 # Variable-length string
# Binary and blob types
TINY_BLOB: int = 249 # TINYBLOB (up to 255 bytes)
MEDIUM_BLOB: int = 250 # MEDIUMBLOB (up to 16MB)
LONG_BLOB: int = 251 # LONGBLOB (up to 4GB)
BLOB: int = 252 # BLOB (up to 64KB)
# Special types
SET: int = 248 # SET (multiple choice from predefined list)
ENUM: int = 247 # ENUM (single choice from predefined list)
GEOMETRY: int = 255 # Spatial data type
JSON: int = 245 # JSON data type (MySQL 5.7+)
NULL: int = 6 # NULL type
@classmethod
def get_string_types(cls) -> List[int]:
"""Get the list of all string types."""
pass
@classmethod
def get_binary_types(cls) -> List[int]:
"""Get the list of all binary types."""
pass
@classmethod
def get_number_types(cls) -> List[int]:
"""Get the list of all number types."""
pass
@classmethod
def get_timestamp_types(cls) -> List[int]:
"""Get the list of all timestamp types."""
passclass FieldFlag:
"""
MySQL field flag constants for column attributes.
Used to identify column properties and constraints.
"""
# Column constraints
NOT_NULL: int = 1 # Column cannot be NULL
PRI_KEY: int = 2 # Column is part of primary key
UNIQUE_KEY: int = 4 # Column is part of unique key
MULTIPLE_KEY: int = 8 # Column is part of non-unique key
# Data characteristics
BLOB: int = 16 # Column is a BLOB or TEXT
UNSIGNED: int = 32 # Numeric column is UNSIGNED
ZEROFILL: int = 64 # Numeric column uses ZEROFILL
BINARY: int = 128 # Column uses BINARY collation
# Special attributes
ENUM: int = 256 # Column is an ENUM
AUTO_INCREMENT: int = 512 # Column has AUTO_INCREMENT
TIMESTAMP: int = 1024 # Column is a TIMESTAMP
SET: int = 2048 # Column is a SET
NO_DEFAULT_VALUE: int = 4096 # Column has no default value
ON_UPDATE_NOW: int = 8192 # Column updates to CURRENT_TIMESTAMP
PART_KEY: int = 16384 # Column is part of some key
GROUP: int = 32768 # Column is part of GROUP BY
UNIQUE: int = 65536 # Column is unique
BINCMP: int = 131072 # Column uses binary comparisonclass ClientFlag:
"""
MySQL client capability flags for connection negotiation.
Indicates client capabilities to the MySQL server.
"""
LONG_PASSWD: int = 1 # Use improved version of old password hashing
FOUND_ROWS: int = 2 # Return found rows instead of affected rows
LONG_FLAG: int = 4 # Get all column flags
CONNECT_WITH_DB: int = 8 # Database name in connection packet
NO_SCHEMA: int = 16 # Don't allow database.table.column
COMPRESS: int = 32 # Use compression protocol
ODBC: int = 64 # ODBC client
LOCAL_FILES: int = 128 # Can use LOAD DATA LOCAL INFILE
IGNORE_SPACE: int = 256 # Ignore spaces before '('
PROTOCOL_41: int = 512 # New 4.1 protocol
INTERACTIVE: int = 1024 # Interactive client
SSL: int = 2048 # Use SSL encryption
IGNORE_SIGPIPE: int = 4096 # Ignore SIGPIPE
TRANSACTIONS: int = 8192 # Client knows about transactions
RESERVED: int = 16384 # Reserved for future use
SECURE_CONNECTION: int = 32768 # New 4.1 authentication
MULTI_STATEMENTS: int = 65536 # Enable multi-statement support
MULTI_RESULTS: int = 131072 # Enable multi-result support
PS_MULTI_RESULTS: int = 262144 # Multi-results in prepared statements
PLUGIN_AUTH: int = 524288 # Client supports plugin authentication
CONNECT_ATTRS: int = 1048576 # Client supports connection attributes
PLUGIN_AUTH_LENENC_CLIENT_DATA: int = 2097152 # Length encoded client data
CAN_HANDLE_EXPIRED_PASSWORDS: int = 4194304 # Handle expired passwords
SESSION_TRACK: int = 8388608 # Session tracking capability
DEPRECATE_EOF: int = 16777216 # Client expects OK packet instead of EOFclass ServerFlag:
"""
MySQL server status flags indicating server state.
Returned in response packets to indicate server status.
"""
SERVER_STATUS_IN_TRANS: int = 1 # Transaction is active
SERVER_STATUS_AUTOCOMMIT: int = 2 # Autocommit is enabled
SERVER_MORE_RESULTS_EXISTS: int = 8 # More results exist (multi-statement)
SERVER_QUERY_NO_GOOD_INDEX_USED: int = 16 # No good index was used
SERVER_QUERY_NO_INDEX_USED: int = 32 # No index was used
SERVER_STATUS_CURSOR_EXISTS: int = 64 # Cursor exists (prepared statements)
SERVER_STATUS_LAST_ROW_SENT: int = 128 # Last row of result sent
SERVER_STATUS_DB_DROPPED: int = 256 # Database was dropped
SERVER_STATUS_NO_BACKSLASH_ESCAPES: int = 512 # NO_BACKSLASH_ESCAPES mode
SERVER_STATUS_METADATA_CHANGED: int = 1024 # Metadata changed
SERVER_QUERY_WAS_SLOW: int = 2048 # Query was slow
SERVER_PS_OUT_PARAMS: int = 4096 # Prepared statement has OUT parameters
SERVER_STATUS_IN_TRANS_READONLY: int = 8192 # In read-only transaction
SERVER_SESSION_STATE_CHANGED: int = 16384 # Session state changedclass ServerCmd:
"""
MySQL server command constants for client-server protocol.
Used in low-level protocol communication.
"""
SLEEP: int = 0 # Internal server thread state
QUIT: int = 1 # Close connection
INIT_DB: int = 2 # Change default database
QUERY: int = 3 # Execute SQL statement
FIELD_LIST: int = 4 # Get column definitions
CREATE_DB: int = 5 # Create database
DROP_DB: int = 6 # Drop database
REFRESH: int = 7 # Refresh/flush tables and logs
SHUTDOWN: int = 8 # Shutdown server
STATISTICS: int = 9 # Get server statistics
PROCESS_INFO: int = 10 # Get process list
CONNECT: int = 11 # Internal server thread state
PROCESS_KILL: int = 12 # Kill server process
DEBUG: int = 13 # Get debug information
PING: int = 14 # Ping server
TIME: int = 15 # Internal server thread state
DELAYED_INSERT: int = 16 # Internal server thread state
CHANGE_USER: int = 17 # Change user authentication
BINLOG_DUMP: int = 18 # Binary log dump (replication)
TABLE_DUMP: int = 19 # Table dump
CONNECT_OUT: int = 20 # Internal server thread state
REGISTER_SLAVE: int = 21 # Register as replication slave
STMT_PREPARE: int = 22 # Prepare statement
STMT_EXECUTE: int = 23 # Execute prepared statement
STMT_SEND_LONG_DATA: int = 24 # Send long data for prepared statement
STMT_CLOSE: int = 25 # Close prepared statement
STMT_RESET: int = 26 # Reset prepared statement
SET_OPTION: int = 27 # Set connection option
STMT_FETCH: int = 28 # Fetch from prepared statement cursor
DAEMON: int = 29 # Internal server thread state
BINLOG_DUMP_GTID: int = 30 # Binary log dump with GTID
RESET_CONNECTION: int = 31 # Reset connection stateclass CharacterSet:
"""
MySQL character set utilities and constants.
Provides character set information and conversion utilities.
"""
# Common character sets
LATIN1_SWEDISH_CI: int = 8 # latin1_swedish_ci (default for latin1)
UTF8_GENERAL_CI: int = 33 # utf8_general_ci (default for utf8)
UTF8_UNICODE_CI: int = 192 # utf8_unicode_ci
UTF8MB4_GENERAL_CI: int = 45 # utf8mb4_general_ci (default for utf8mb4)
UTF8MB4_UNICODE_CI: int = 224 # utf8mb4_unicode_ci
UTF8MB4_0900_AI_CI: int = 255 # utf8mb4_0900_ai_ci (MySQL 8.0 default)
BINARY: int = 63 # binary
@classmethod
def get_info(cls, charset_id: int) -> Dict[str, Union[str, int]]:
"""
Get character set information by ID.
Args:
charset_id: MySQL character set ID
Returns:
Dictionary with charset info (name, collation, etc.)
"""
pass
@classmethod
def get_default_collation(cls, charset: str) -> Tuple[int, str]:
"""
Get default collation for character set.
Args:
charset: Character set name
Returns:
Tuple of (collation_id, collation_name)
"""
pass
@classmethod
def get_charset_info(cls, charset: str) -> Dict[str, Any]:
"""
Get comprehensive character set information.
Args:
charset: Character set name
Returns:
Dictionary with detailed charset information
"""
passclass RefreshOption:
"""
MySQL refresh option constants for FLUSH operations.
Used with CMD_REFRESH command and FLUSH statements.
"""
GRANT: int = 1 # Reload grant tables
LOG: int = 2 # Flush logs
TABLES: int = 4 # Close all open tables
HOSTS: int = 8 # Flush host cache
STATUS: int = 16 # Reset status variables
THREADS: int = 32 # Flush thread cache
SLAVE: int = 64 # Reset replica (slave) info
MASTER: int = 128 # Remove binary logs listed in index
ERROR_LOG: int = 256 # Flush error log
ENGINE_LOG: int = 512 # Flush engine logs
BINARY_LOG: int = 1024 # Flush binary log
RELAY_LOG: int = 2048 # Flush relay log
GENERAL_LOG: int = 4096 # Flush general log
SLOW_LOG: int = 8192 # Flush slow query log# SSL mode constants
SSL_DISABLED: int = 0 # SSL connections disabled
SSL_PREFERRED: int = 1 # SSL preferred but not required
SSL_REQUIRED: int = 2 # SSL required
SSL_VERIFY_CA: int = 3 # SSL required with CA verification
SSL_VERIFY_IDENTITY: int = 4 # SSL required with full verification
# TLS version constants
TLS_VERSIONS: Dict[str, str] = {
'TLSv1': 'TLSv1',
'TLSv1.1': 'TLSv1.1',
'TLSv1.2': 'TLSv1.2',
'TLSv1.3': 'TLSv1.3'
}# TLS cipher suite definitions from mysql.connector.tls_ciphers
TLS_CIPHER_SUITES: Dict[str, List[str]] = {
'TLSv1.2': [
'ECDHE-RSA-AES128-GCM-SHA256',
'ECDHE-RSA-AES256-GCM-SHA384',
'ECDHE-ECDSA-AES128-GCM-SHA256',
'ECDHE-ECDSA-AES256-GCM-SHA384',
'DHE-RSA-AES128-GCM-SHA256',
'DHE-RSA-AES256-GCM-SHA384',
# ... additional cipher suites
],
'TLSv1.3': [
'TLS_AES_256_GCM_SHA384',
'TLS_CHACHA20_POLY1305_SHA256',
'TLS_AES_128_GCM_SHA256',
# ... additional TLS 1.3 cipher suites
]
}# Default connection configuration values
DEFAULT_CONFIGURATION: Dict[str, Any] = {
'host': 'localhost',
'port': 3306,
'user': '',
'password': '',
'database': '',
'charset': 'utf8mb4',
'collation': 'utf8mb4_general_ci',
'use_unicode': True,
'autocommit': False,
'time_zone': None,
'sql_mode': None,
'connect_timeout': 10,
'read_timeout': None,
'write_timeout': None,
'client_flags': 0,
'compress': False,
'buffered': False,
'raw': False,
'consume_results': False,
'ssl_disabled': False,
'ssl_verify_cert': False,
'ssl_verify_identity': False,
'force_ipv6': False,
'use_pure': None,
'auth_plugin': None,
'pool_name': None,
'pool_size': 5,
'pool_reset_session': True,
'pool_timeout': 0,
'allow_local_infile': True,
'allow_local_infile_in_path': None,
'get_warnings': False,
'raise_on_warnings': False,
'prepared': False,
'converter_class': None,
'converter_str_fallback': False,
'failover': [],
'option_files': None,
'option_groups': ['client', 'mysql_connector_python'],
}import mysql.connector
from mysql.connector.constants import FieldType
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
cursor.execute("DESCRIBE users")
# Check field types in table structure
for row in cursor:
field_name = row[0]
field_type = row[1]
# Parse field type (simplified example)
if 'int' in field_type.lower():
print(f"{field_name}: Integer type")
elif 'varchar' in field_type.lower():
print(f"{field_name}: Variable string type")
elif 'datetime' in field_type.lower():
print(f"{field_name}: Datetime type")
elif 'text' in field_type.lower():
print(f"{field_name}: Text type")
cursor.close()
connection.close()import mysql.connector
from mysql.connector.constants import FieldFlag
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
cursor.execute("SELECT * FROM users LIMIT 1")
# Analyze field flags from cursor description
for i, column_desc in enumerate(cursor.description):
name, type_code, display_size, internal_size, precision, scale, null_ok = column_desc
# Note: Field flags would be available in extended cursor descriptions
# This is a conceptual example
print(f"Column {name}:")
print(f" Type: {type_code}")
print(f" Nullable: {null_ok}")
print(f" Display size: {display_size}")
cursor.close()
connection.close()import mysql.connector
from mysql.connector.constants import ClientFlag
# Configure specific client capabilities
client_flags = (
ClientFlag.FOUND_ROWS | # Return found rows, not affected rows
ClientFlag.LONG_FLAG | # Get all column flags
ClientFlag.CONNECT_WITH_DB | # Send database name in connection
ClientFlag.COMPRESS | # Use compression
ClientFlag.LOCAL_FILES | # Allow LOAD DATA LOCAL INFILE
ClientFlag.MULTI_STATEMENTS | # Enable multi-statement support
ClientFlag.MULTI_RESULTS | # Enable multi-result support
ClientFlag.SSL # Use SSL encryption
)
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase',
client_flags=client_flags
)
# Connection now uses specified client capabilities
print(f"Connected with client flags: {client_flags}")
connection.close()import mysql.connector
from mysql.connector.constants import CharacterSet
# Get character set information
charset_info = CharacterSet.get_charset_info('utf8mb4')
print(f"Character set info: {charset_info}")
# Get default collation
collation_id, collation_name = CharacterSet.get_default_collation('utf8mb4')
print(f"Default collation: {collation_name} (ID: {collation_id})")
# Connect with specific character set
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase',
charset='utf8mb4',
collation='utf8mb4_unicode_ci'
)
print(f"Connection charset: {connection.charset}")
print(f"Connection collation: {connection.collation}")
connection.close()import mysql.connector
from mysql.connector.constants import ServerFlag
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
cursor = connection.cursor()
# Execute query and check server status
cursor.execute("SELECT COUNT(*) FROM users")
result = cursor.fetchone()
# Note: Server status flags would be available in result metadata
# This is a conceptual example of how you might check transaction status
if connection.in_transaction:
print("Currently in a transaction")
else:
print("Not in a transaction")
if connection.autocommit:
print("Autocommit is enabled")
else:
print("Autocommit is disabled")
cursor.close()
connection.close()import mysql.connector
from mysql.connector.constants import RefreshOption
connection = mysql.connector.connect(
host='localhost',
user='myuser',
password='mypassword',
database='mydatabase'
)
# Flush various server caches (requires appropriate privileges)
refresh_options = RefreshOption.TABLES | RefreshOption.STATUS
try:
# Send refresh command (low-level protocol operation)
result = connection.cmd_refresh(refresh_options)
print("Server caches refreshed successfully")
except mysql.connector.Error as err:
print(f"Refresh failed: {err}")
connection.close()import mysql.connector
# SSL configuration using constants
ssl_config = {
'ssl_disabled': False,
'ssl_verify_cert': True,
'ssl_verify_identity': True,
'ssl_cert': '/path/to/client-cert.pem',
'ssl_key': '/path/to/client-key.pem',
'ssl_ca': '/path/to/ca-cert.pem',
'tls_versions': ['TLSv1.2', 'TLSv1.3'],
'tls_ciphersuites': [
'TLS_AES_256_GCM_SHA384',
'TLS_CHACHA20_POLY1305_SHA256'
]
}
try:
connection = mysql.connector.connect(
host='mysql.example.com',
user='myuser',
password='mypassword',
database='mydatabase',
**ssl_config
)
print("Secure SSL/TLS connection established")
connection.close()
except mysql.connector.Error as err:
print(f"SSL connection failed: {err}")Install with Tessl CLI
npx tessl i tessl/pypi-mysql-connector-python