tessl/pypi-mysql-connector
MySQL driver written in Python providing comprehensive database connectivity with both traditional SQL and modern document operations
—
Pending
django-integration.mddocs/
Django Integration
Complete Django ORM backend providing seamless integration with Django applications, including database introspection, feature detection, and operation optimization.
Capabilities
Database Backend
Django database backend implementation for MySQL Connector Python.
class DatabaseWrapper:
"""
Django database backend wrapper for MySQL Connector Python.
Provides complete Django ORM integration with MySQL databases,
supporting all Django database features including migrations,
transactions, and query optimization.
"""
def __init__(self, settings_dict, alias='default'):
"""
Initialize database wrapper.
Parameters:
- settings_dict (dict): Database configuration from Django settings
- alias (str): Database alias name
"""
def get_connection_params(self):
"""
Get connection parameters from Django settings.
Returns:
dict: MySQL Connector connection parameters
"""
def get_new_connection(self, conn_params):
"""
Create new database connection.
Parameters:
- conn_params (dict): Connection parameters
Returns:
MySQLConnection: Database connection object
"""
def init_connection_state(self):
"""Initialize connection state for Django requirements"""
def create_cursor(self, name=None):
"""
Create database cursor.
Parameters:
- name (str): Cursor name (unused)
Returns:
CursorWrapper: Django cursor wrapper
"""
def _set_autocommit(self, autocommit):
"""Set connection autocommit mode"""
def disable_constraint_checking(self):
"""Disable foreign key constraint checking"""
def enable_constraint_checking(self):
"""Enable foreign key constraint checking"""
def check_constraints(self, table_names=None):
"""Check integrity constraints"""
def is_usable(self):
"""Check if connection is still usable"""
@property
def queries_limit(self):
"""Query limit for debug logging"""
@property
def queries_logged(self):
"""Number of queries logged"""Database Features
Detection and configuration of MySQL-specific database features.
class DatabaseFeatures:
"""
Database feature detection and configuration for MySQL.
Defines which Django features are supported by MySQL and
provides MySQL-specific behavior configuration.
"""
# Feature flags
empty_fetchmany_value = ()
update_can_self_select = False
allows_group_by_pk = True
related_fields_match_type = True
allow_sliced_subqueries = False
has_bulk_insert = True
has_select_for_update = True
has_select_for_update_nowait = False
supports_forward_references = True
supports_long_model_names = False
supports_microsecond_precision = False
supports_regex_backreferencing = False
supports_date_lookup_using_string = False
can_introspect_autofield = True
can_introspect_binary_field = False
can_introspect_boolean_field = False
can_introspect_null = True
can_introspect_positive_integer_field = True
can_introspect_small_integer_field = True
supports_timezones = False
requires_explicit_null_ordering_when_grouping = True
allows_auto_pk_0 = False
uses_savepoints = True
atomic_transactions = False
def __init__(self, connection):
"""Initialize with database connection"""
self.connection = connection
@cached_property
def supports_transactions(self):
"""Check if MySQL storage engine supports transactions"""
@cached_property
def has_zoneinfo_database(self):
"""Check if MySQL has timezone database loaded"""
@cached_property
def supports_over_clause(self):
"""Check if MySQL version supports window functions"""
@cached_property
def supports_frame_range_fixed_distance(self):
"""Check window function frame support"""Database Operations
MySQL-specific database operations and SQL generation.
class DatabaseOperations:
"""
MySQL-specific database operations and SQL generation.
Provides MySQL-specific implementations of database operations
required by Django ORM, including SQL quoting, date formatting,
and aggregate functions.
"""
compiler_module = "mysql.connector.django.compiler"
def __init__(self, connection):
"""Initialize with database connection"""
self.connection = connection
def autoinc_sql(self, table, column):
"""
Generate auto-increment SQL for column.
Parameters:
- table (str): Table name
- column (str): Column name
Returns:
str: AUTO_INCREMENT SQL
"""
return "AUTO_INCREMENT"
def bulk_batch_size(self, fields, objs):
"""
Calculate optimal batch size for bulk operations.
Parameters:
- fields (list): Field list
- objs (list): Object list
Returns:
int: Optimal batch size
"""
def quote_name(self, name):
"""
Quote database identifier name.
Parameters:
- name (str): Database identifier
Returns:
str: Quoted identifier
"""
return "`%s`" % name.replace("`", "``")
def random_function_sql(self):
"""Get SQL for random number generation"""
return "RAND()"
def sql_flush(self, style, tables, sequences, allow_cascade=False):
"""
Generate SQL to flush database tables.
Parameters:
- style: SQL style formatter
- tables (list): Table names to flush
- sequences (list): Sequence names to reset
- allow_cascade (bool): Allow cascading deletes
Returns:
list: SQL statements
"""
def validate_autopk_value(self, value):
"""Validate auto primary key value"""
return 0 < value < 18446744073709551616
def adapt_datetimefield_value(self, value):
"""Adapt datetime value for MySQL storage"""
def adapt_timefield_value(self, value):
"""Adapt time value for MySQL storage"""
def date_extract_sql(self, lookup_type, field_name):
"""Generate SQL for date extraction"""
def date_trunc_sql(self, lookup_type, field_name):
"""Generate SQL for date truncation"""
def datetime_extract_sql(self, lookup_type, field_name, tzname):
"""Generate SQL for datetime extraction"""
def datetime_trunc_sql(self, lookup_type, field_name, tzname):
"""Generate SQL for datetime truncation"""
def time_extract_sql(self, lookup_type, field_name):
"""Generate SQL for time extraction"""
def force_no_ordering(self):
"""Get SQL to force no result ordering"""
return [(None, ("NULL", [], False))]
def fulltext_search_sql(self, field_name):
"""Generate SQL for full-text search"""
return "MATCH (%s) AGAINST (%%s IN BOOLEAN MODE)" % field_name
def last_insert_id(self, cursor, table_name, pk_name):
"""Get last inserted row ID"""
return cursor.lastrowid
def lookup_cast(self, lookup_type, internal_type=None):
"""Get SQL cast for field lookup"""
def max_name_length(self):
"""Maximum identifier name length"""
return 64
def pk_default_value(self):
"""Default primary key value"""
return "NULL"
def prep_for_iexact_query(self, x):
"""Prepare value for case-insensitive exact lookup"""
return x
def process_clob(self, value):
"""Process CLOB field value"""
return value
def regex_lookup(self, lookup_type):
"""Get SQL operator for regex lookup"""
if lookup_type == 'regex':
return '%s REGEXP BINARY %s'
else:
return '%s REGEXP %s'
def savepoint_create_sql(self, sid):
"""Generate SQL to create savepoint"""
return "SAVEPOINT %s" % self.quote_name(sid)
def savepoint_rollback_sql(self, sid):
"""Generate SQL to rollback to savepoint"""
return "ROLLBACK TO SAVEPOINT %s" % self.quote_name(sid)
def savepoint_commit_sql(self, sid):
"""Generate SQL to release savepoint"""
return "RELEASE SAVEPOINT %s" % self.quote_name(sid)Database Introspection
Schema introspection capabilities for Django migrations and admin.
class DatabaseIntrospection:
"""
Database schema introspection for Django migrations and admin.
Provides methods to inspect existing database schema and
extract metadata for Django model generation and migrations.
"""
data_types_reverse = {
FIELD_TYPE.BLOB: 'TextField',
FIELD_TYPE.CHAR: 'CharField',
FIELD_TYPE.DECIMAL: 'DecimalField',
FIELD_TYPE.NEWDECIMAL: 'DecimalField',
FIELD_TYPE.DATE: 'DateField',
FIELD_TYPE.DATETIME: 'DateTimeField',
FIELD_TYPE.DOUBLE: 'FloatField',
FIELD_TYPE.FLOAT: 'FloatField',
FIELD_TYPE.INT24: 'IntegerField',
FIELD_TYPE.LONG: 'IntegerField',
FIELD_TYPE.LONGLONG: 'BigIntegerField',
FIELD_TYPE.SHORT: 'SmallIntegerField',
FIELD_TYPE.STRING: 'CharField',
FIELD_TYPE.TIMESTAMP: 'DateTimeField',
FIELD_TYPE.TINY: 'IntegerField',
FIELD_TYPE.TINY_BLOB: 'TextField',
FIELD_TYPE.MEDIUM_BLOB: 'TextField',
FIELD_TYPE.LONG_BLOB: 'TextField',
FIELD_TYPE.VAR_STRING: 'CharField',
FIELD_TYPE.VARCHAR: 'CharField',
FIELD_TYPE.TIME: 'TimeField',
FIELD_TYPE.YEAR: 'IntegerField',
}
def __init__(self, connection):
"""Initialize with database connection"""
self.connection = connection
def get_table_list(self, cursor):
"""
Get list of tables in database.
Parameters:
- cursor: Database cursor
Returns:
list: Table information tuples
"""
def get_table_description(self, cursor, table_name):
"""
Get column descriptions for table.
Parameters:
- cursor: Database cursor
- table_name (str): Table name
Returns:
list: Column description tuples
"""
def get_relations(self, cursor, table_name):
"""
Get foreign key relations for table.
Parameters:
- cursor: Database cursor
- table_name (str): Table name
Returns:
dict: Column to referenced table/column mapping
"""
def get_key_columns(self, cursor, table_name):
"""
Get foreign key column information.
Parameters:
- cursor: Database cursor
- table_name (str): Table name
Returns:
list: Foreign key column information
"""
def get_indexes(self, cursor, table_name):
"""
Get index information for table.
Parameters:
- cursor: Database cursor
- table_name (str): Table name
Returns:
dict: Index information mapping
"""
def get_constraints(self, cursor, table_name):
"""
Get constraint information for table.
Parameters:
- cursor: Database cursor
- table_name (str): Table name
Returns:
dict: Constraint information mapping
"""
def data_type_check_constraints(self, field_type):
"""Get check constraints for field type"""
def describe_table(self, cursor, table_name):
"""Get detailed table description"""Django Settings Configuration
Django settings configuration for MySQL Connector Python backend.
# Django settings.py configuration example
DATABASES = {
'default': {
'ENGINE': 'mysql.connector.django',
'NAME': 'django_db',
'USER': 'django_user',
'PASSWORD': 'django_password',
'HOST': 'localhost',
'PORT': 3306,
'OPTIONS': {
'autocommit': True,
'init_command': "SET sql_mode='STRICT_TRANS_TABLES'",
'charset': 'utf8mb4',
'use_unicode': True,
'sql_mode': 'STRICT_TRANS_TABLES',
'isolation_level': 'read committed',
},
}
}Usage Example:
# Django model using MySQL-specific features
from django.db import models
class User(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)
created_at = models.DateTimeField(auto_now_add=True)
bio = models.TextField(blank=True)
class Meta:
db_table = 'users'
indexes = [
models.Index(fields=['email']),
models.Index(fields=['created_at']),
]
# Django ORM queries work seamlessly
users = User.objects.filter(created_at__year=2023)
User.objects.bulk_create([
User(name='John', email='john@example.com'),
User(name='Jane', email='jane@example.com'),
])
# Raw SQL with Django
from django.db import connection
with connection.cursor() as cursor:
cursor.execute("SELECT COUNT(*) FROM users WHERE created_at > %s", ['2023-01-01'])
count = cursor.fetchone()[0]Types
DjangoSettings = {
'ENGINE': str,
'NAME': str,
'USER': str,
'PASSWORD': str,
'HOST': str,
'PORT': int,
'OPTIONS': dict[str, any]
}
TableInfo = tuple[str, str] # (table_name, table_type)
ColumnDescription = tuple[
str, # column_name
str, # data_type
int, # display_size
int, # internal_size
int, # precision
int, # scale
bool # null_ok
]
IndexInfo = {
'columns': list[str],
'primary_key': bool,
'unique': bool,
'foreign_key': tuple[str, str] or None,
'check': bool,
'type': str,
'orders': list[str]
}Install with Tessl CLI
npx tessl i tessl/pypi-mysql-connector