MySQL driver written in Python providing comprehensive database connectivity with both traditional SQL and modern document operations
npx @tessl/cli install tessl/pypi-mysql-connector@2.2.0A comprehensive MySQL database driver written in pure Python, providing complete connectivity for both traditional SQL operations and modern document-based operations. The package offers two complementary interfaces: the traditional mysql.connector module with full DB-API 2.0 compliance, and the modern mysqlx module supporting MySQL's X DevAPI for document and relational data.
pip install mysql-connectorStandard MySQL connectivity:
import mysql.connectorMySQL X DevAPI for document operations:
import mysqlximport mysql.connector
# Connect to MySQL database
config = {
'user': 'username',
'password': 'password',
'host': 'localhost',
'database': 'mydb',
'port': 3306
}
connection = mysql.connector.connect(**config)
cursor = connection.cursor()
# Execute SQL query
cursor.execute("SELECT * FROM users WHERE age > %s", (18,))
results = cursor.fetchall()
for row in results:
print(row)
cursor.close()
connection.close()import mysqlx
# Connect using X Protocol
session = mysqlx.get_session({
'host': 'localhost',
'port': 33060,
'user': 'username',
'password': 'password'
})
# Work with collections (document store)
schema = session.get_schema('mydb')
collection = schema.get_collection('users')
# Add documents
collection.add({'name': 'John', 'age': 25, 'email': 'john@example.com'}).execute()
# Find documents
docs = collection.find('age > :age').bind('age', 18).execute()
for doc in docs:
print(doc)
session.close()The package provides two distinct yet complementary APIs:
Both modules share common error handling patterns and support advanced features like connection pooling, SSL/TLS encryption, authentication plugins, and comprehensive charset support.
Core connection functionality with support for connection pooling, failover configurations, SSL/TLS encryption, and various authentication methods including plugin-based authentication.
def connect(**config):
"""Create MySQL connection with configuration options"""
class MySQLConnection:
"""Pure Python MySQL connection implementation"""
class MySQLConnectionPool:
"""Connection pool for managing multiple connections"""Traditional SQL operations with multiple cursor types for different data access patterns, including dictionary cursors, buffered cursors, prepared statements, and raw data access.
class MySQLCursor:
"""Standard cursor for SQL operations"""
def execute(self, statement, params=None): ...
def fetchone(self): ...
def fetchmany(self, size=1): ...
def fetchall(self): ...
class MySQLCursorDict:
"""Dictionary-based cursor returning rows as dicts"""Modern document database operations using MySQL's X DevAPI, supporting JSON document storage, retrieval, and manipulation with a fluent API design.
def get_session(connection_data):
"""Create X Protocol session for document operations"""
class Collection:
"""Document collection operations"""
def add(self, *docs): ...
def find(self, condition=None): ...
def modify(self, condition): ...
def remove(self, condition): ...Comprehensive exception hierarchy following DB-API 2.0 standards, with specific error types for different failure scenarios and custom error handling capabilities.
class Error(Exception):
"""Base exception class for all MySQL errors"""
class DatabaseError(Error):
"""Database-related error exceptions"""
class OperationalError(DatabaseError):
"""Database operation error exceptions"""Complete Django ORM backend providing seamless integration with Django applications, including database introspection, feature detection, and operation optimization.
class DatabaseWrapper:
"""Django database backend wrapper"""
class DatabaseOperations:
"""Django-specific database operations"""Integration with MySQL Fabric for high availability and scaling, supporting automatic failover, load balancing, and sharding configurations.
def connect(**config):
"""Create Fabric-aware MySQL connection"""
class MySQLFabricConnection:
"""MySQL Fabric connection management"""# DB-API 2.0 Module Attributes
apilevel: str = '2.0'
threadsafety: int = 1
paramstyle: str = 'pyformat'
# Type Constructors
def Date(year: int, month: int, day: int) -> datetime.date:
"""Construct date object from year, month, day"""
def Time(hour: int, minute: int, second: int) -> datetime.time:
"""Construct time object from hour, minute, second"""
def Timestamp(year: int, month: int, day: int, hour: int, minute: int, second: int) -> datetime.datetime:
"""Construct timestamp object from date and time components"""
def DateFromTicks(ticks: float) -> datetime.date:
"""Construct date object from timestamp in seconds since epoch"""
def TimeFromTicks(ticks: float) -> datetime.time:
"""Construct time object from timestamp in seconds since epoch"""
def TimestampFromTicks(ticks: float) -> datetime.datetime:
"""Construct timestamp object from timestamp in seconds since epoch"""
def Binary(data: bytes) -> bytes:
"""Construct binary object for binary data"""
# Type Objects for comparisons
STRING: _DBAPITypeObject
BINARY: _DBAPITypeObject
NUMBER: _DBAPITypeObject
DATETIME: _DBAPITypeObject
ROWID: _DBAPITypeObjectConnectionConfig = {
'user': str,
'password': str,
'host': str,
'port': int,
'database': str,
'charset': str,
'use_pure': bool,
'ssl_disabled': bool,
'ssl_cert': str,
'ssl_key': str,
'ssl_ca': str,
'pool_name': str,
'pool_size': int,
'pool_reset_session': bool,
'autocommit': bool,
'time_zone': str,
'sql_mode': str,
'init_command': str,
'read_timeout': int,
'write_timeout': int,
'connect_timeout': int,
'compress': bool,
'auth_plugin': str,
'unix_socket': str,
'use_unicode': bool,
'collation': str
}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
NEWDATE: int
VARCHAR: int
BIT: int
JSON: int
NEWDECIMAL: int
ENUM: int
SET: int
TINY_BLOB: int
MEDIUM_BLOB: int
LONG_BLOB: int
BLOB: int
VAR_STRING: int
STRING: int
GEOMETRY: int
class FieldFlag:
"""MySQL field attribute flags"""
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"""
LONG_PASSWD: int
FOUND_ROWS: int
LONG_FLAG: int
CONNECT_WITH_DB: int
NO_SCHEMA: int
COMPRESS: int
ODBC: int
LOCAL_FILES: int
IGNORE_SPACE: int
PROTOCOL_41: int
INTERACTIVE: int
SSL: int
IGNORE_SIGPIPE: int
TRANSACTIONS: int
RESERVED: int
SECURE_CONNECTION: int
MULTI_STATEMENTS: int
MULTI_RESULTS: int
PS_MULTI_RESULTS: int
PLUGIN_AUTH: int
CONNECT_ARGS: int
PLUGIN_AUTH_LENENC_CLIENT_DATA: int
CAN_HANDLE_EXPIRED_PASSWORDS: int
SESSION_TRACK: int
DEPRECATE_EOF: int
class ServerFlag:
"""MySQL server status flags"""
STATUS_IN_TRANS: int
STATUS_AUTOCOMMIT: int
STATUS_MORE_RESULTS_EXISTS: int
STATUS_NO_GOOD_INDEX_USED: int
STATUS_NO_INDEX_USED: int
STATUS_CURSOR_EXISTS: int
STATUS_LAST_ROW_SENT: int
STATUS_DB_DROPPED: int
STATUS_NO_BACKSLASH_ESCAPES: int
STATUS_METADATA_CHANGED: int
STATUS_QUERY_WAS_SLOW: int
STATUS_PS_OUT_PARAMS: int
class RefreshOption:
"""MySQL FLUSH/REFRESH options"""
GRANT: int
LOG: int
TABLES: int
HOSTS: int
STATUS: int
THREADS: int
SLAVE: int
MASTER: int
class CharacterSet:
"""MySQL character set constants"""
LATIN1: int
UTF8: int
UTF8MB4: int
BINARY: int