import uuid from django.conf import settings from django.db.backends.base.operations import BaseDatabaseOperations from django.db.backends.utils import split_tzname_delta from django.db.models import Exists, ExpressionWrapper, Lookup from django.db.models.constants import OnConflict from django.utils import timezone from django.utils.encoding import force_str from django.utils.regex_helper import _lazy_re_compile class DatabaseOperations(BaseDatabaseOperations): compiler_module = "django.db.backends.mysql.compiler" # MySQL stores positive fields as UNSIGNED ints. integer_field_ranges = { **BaseDatabaseOperations.integer_field_ranges, "PositiveSmallIntegerField": (0, 65535), "PositiveIntegerField": (0, 4294967295), "PositiveBigIntegerField": (0, 18446744073709551615), } cast_data_types = { "AutoField": "signed integer", "BigAutoField": "signed integer", "SmallAutoField": "signed integer", "CharField": "char(%(max_length)s)", "DecimalField": "decimal(%(max_digits)s, %(decimal_places)s)", "TextField": "char", "IntegerField": "signed integer", "BigIntegerField": "signed integer", "SmallIntegerField": "signed integer", "PositiveBigIntegerField": "unsigned integer", "PositiveIntegerField": "unsigned integer", "PositiveSmallIntegerField": "unsigned integer", "DurationField": "signed integer", } cast_char_field_without_max_length = "char" explain_prefix = "EXPLAIN" # EXTRACT format cannot be passed in parameters. _extract_format_re = _lazy_re_compile(r"[A-Z_]+") def date_extract_sql(self, lookup_type, sql, params): # https://dev.mysql.com/doc/mysql/en/date-and-time-functions.html if lookup_type == "week_day": # DAYOFWEEK() returns an integer, 1-7, Sunday=1. return f"DAYOFWEEK({sql})", params elif lookup_type == "iso_week_day": # WEEKDAY() returns an integer, 0-6, Monday=0. return f"WEEKDAY({sql}) + 1", params elif lookup_type == "week": # Override the value of default_week_format for consistency with # other database backends. # Mode 3: Monday, 1-53, with 4 or more days this year. return f"WEEK({sql}, 3)", params elif lookup_type == "iso_year": # Get the year part from the YEARWEEK function, which returns a # number as year * 100 + week. return f"TRUNCATE(YEARWEEK({sql}, 3), -2) / 100", params else: # EXTRACT returns 1-53 based on ISO-8601 for the week number. lookup_type = lookup_type.upper() if not self._extract_format_re.fullmatch(lookup_type): raise ValueError(f"Invalid loookup type: {lookup_type!r}") return f"EXTRACT({lookup_type} FROM {sql})", params def date_trunc_sql(self, lookup_type, sql, params, tzname=None): sql, params = self._convert_sql_to_tz(sql, params, tzname) fields = { "year": "%Y-01-01", "month": "%Y-%m-01", } if lookup_type in fields: format_str = fields[lookup_type] return f"CAST(DATE_FORMAT({sql}, %s) AS DATE)", (*params, format_str) elif lookup_type == "quarter": return ( f"MAKEDATE(YEAR({sql}), 1) + " f"INTERVAL QUARTER({sql}) QUARTER - INTERVAL 1 QUARTER", (*params, *params), ) elif lookup_type == "week": return f"DATE_SUB({sql}, INTERVAL WEEKDAY({sql}) DAY)", (*params, *params) else: return f"DATE({sql})", params def _prepare_tzname_delta(self, tzname): tzname, sign, offset = split_tzname_delta(tzname) return f"{sign}{offset}" if offset else tzname def _convert_sql_to_tz(self, sql, params, tzname): if tzname and settings.USE_TZ and self.connection.timezone_name != tzname: return f"CONVERT_TZ({sql}, %s, %s)", ( *params, self.connection.timezone_name, self._prepare_tzname_delta(tzname), ) return sql, params def datetime_cast_date_sql(self, sql, params, tzname): sql, params = self._convert_sql_to_tz(sql, params, tzname) return f"DATE({sql})", params def datetime_cast_time_sql(self, sql, params, tzname): sql, params = self._convert_sql_to_tz(sql, params, tzname) return f"TIME({sql})", params def datetime_extract_sql(self, lookup_type, sql, params, tzname): sql, params = self._convert_sql_to_tz(sql, params, tzname) return self.date_extract_sql(lookup_type, sql, params) def datetime_trunc_sql(self, lookup_type, sql, params, tzname): sql, params = self._convert_sql_to_tz(sql, params, tzname) fields = ["year", "month", "day", "hour", "minute", "second"] format = ("%Y-", "%m", "-%d", " %H:", "%i", ":%s") format_def = ("0000-", "01", "-01", " 00:", "00", ":00") if lookup_type == "quarter": return ( f"CAST(DATE_FORMAT(MAKEDATE(YEAR({sql}), 1) + " f"INTERVAL QUARTER({sql}) QUARTER - " f"INTERVAL 1 QUARTER, %s) AS DATETIME)" ), (*params, *params, "%Y-%m-01 00:00:00") if lookup_type == "week": return ( f"CAST(DATE_FORMAT(" f"DATE_SUB({sql}, INTERVAL WEEKDAY({sql}) DAY), %s) AS DATETIME)" ), (*params, *params, "%Y-%m-%d 00:00:00") try: i = fields.index(lookup_type) + 1 except ValueError: pass else: format_str = "".join(format[:i] + format_def[i:]) return f"CAST(DATE_FORMAT({sql}, %s) AS DATETIME)", (*params, format_str) return sql, params def time_trunc_sql(self, lookup_type, sql, params, tzname=None): sql, params = self._convert_sql_to_tz(sql, params, tzname) fields = { "hour": "%H:00:00", "minute": "%H:%i:00", "second": "%H:%i:%s", } if lookup_type in fields: format_str = fields[lookup_type] return f"CAST(DATE_FORMAT({sql}, %s) AS TIME)", (*params, format_str) else: return f"TIME({sql})", params def fetch_returned_insert_rows(self, cursor): """ Given a cursor object that has just performed an INSERT...RETURNING statement into a table, return the tuple of returned data. """ return cursor.fetchall() def format_for_duration_arithmetic(self, sql): return "INTERVAL %s MICROSECOND" % sql def force_no_ordering(self): """ "ORDER BY NULL" prevents MySQL from implicitly ordering by grouped columns. If no ordering would otherwise be applied, we don't want any implicit sorting going on. """ return [(None, ("NULL", [], False))] def adapt_decimalfield_value(self, value, max_digits=None, decimal_places=None): return value def last_executed_query(self, cursor, sql, params): # With MySQLdb, cursor objects have an (undocumented) "_executed" # attribute where the exact query sent to the database is saved. # See MySQLdb/cursors.py in the source distribution. # MySQLdb returns string, PyMySQL bytes. return force_str(getattr(cursor, "_executed", None), errors="replace") def no_limit_value(self): # 2**64 - 1, as recommended by the MySQL documentation return 18446744073709551615 def quote_name(self, name): if name.startswith("`") and name.endswith("`"): return name # Quoting once is enough. return "`%s`" % name def return_insert_columns(self, fields): # MySQL and MariaDB < 10.5.0 don't support an INSERT...RETURNING # statement. if not fields: return "", () columns = [ "%s.%s" % ( self.quote_name(field.model._meta.db_table), self.quote_name(field.column), ) for field in fields ] return "RETURNING %s" % ", ".join(columns), () def sql_flush(self, style, tables, *, reset_sequences=False, allow_cascade=False): if not tables: return [] sql = ["SET FOREIGN_KEY_CHECKS = 0;"] if reset_sequences: # It's faster to TRUNCATE tables that require a sequence reset # since ALTER TABLE AUTO_INCREMENT is slower than TRUNCATE. sql.extend( "%s %s;" % ( style.SQL_KEYWORD("TRUNCATE"), style.SQL_FIELD(self.quote_name(table_name)), ) for table_name in tables ) else: # Otherwise issue a simple DELETE since it's faster than TRUNCATE # and preserves sequences. sql.extend( "%s %s %s;" % ( style.SQL_KEYWORD("DELETE"), style.SQL_KEYWORD("FROM"), style.SQL_FIELD(self.quote_name(table_name)), ) for table_name in tables ) sql.append("SET FOREIGN_KEY_CHECKS = 1;") return sql def sequence_reset_by_name_sql(self, style, sequences): return [ "%s %s %s %s = 1;" % ( style.SQL_KEYWORD("ALTER"), style.SQL_KEYWORD("TABLE"), style.SQL_FIELD(self.quote_name(sequence_info["table"])), style.SQL_FIELD("AUTO_INCREMENT"), ) for sequence_info in sequences ] def validate_autopk_value(self, value): # Zero in AUTO_INCREMENT field does not work without the # NO_AUTO_VALUE_ON_ZERO SQL mode. if value == 0 and not self.connection.features.allows_auto_pk_0: raise ValueError( "The database backend does not accept 0 as a value for AutoField." ) return value def adapt_datetimefield_value(self, value): if value is None: return None # Expression values are adapted by the database. if hasattr(value, "resolve_expression"): return value # MySQL doesn't support tz-aware datetimes if timezone.is_aware(value): if settings.USE_TZ: value = timezone.make_naive(value, self.connection.timezone) else: raise ValueError( "MySQL backend does not support timezone-aware datetimes when " "USE_TZ is False." ) return str(value) def adapt_timefield_value(self, value): if value is None: return None # Expression values are adapted by the database. if hasattr(value, "resolve_expression"): return value # MySQL doesn't support tz-aware times if timezone.is_aware(value): raise ValueError("MySQL backend does not support timezone-aware times.") return value.isoformat(timespec="microseconds") def max_name_length(self): return 64 def pk_default_value(self): return "NULL" def bulk_insert_sql(self, fields, placeholder_rows): placeholder_rows_sql = (", ".join(row) for row in placeholder_rows) values_sql = ", ".join("(%s)" % sql for sql in placeholder_rows_sql) return "VALUES " + values_sql def combine_expression(self, connector, sub_expressions): if connector == "^": return "POW(%s)" % ",".join(sub_expressions) # Convert the result to a signed integer since MySQL's binary operators # return an unsigned integer. elif connector in ("&", "|", "<<", "#"): connector = "^" if connector == "#" else connector return "CONVERT(%s, SIGNED)" % connector.join(sub_expressions) elif connector == ">>": lhs, rhs = sub_expressions return "FLOOR(%(lhs)s / POW(2, %(rhs)s))" % {"lhs": lhs, "rhs": rhs} return super().combine_expression(connector, sub_expressions) def get_db_converters(self, expression): converters = super().get_db_converters(expression) internal_type = expression.output_field.get_internal_type() if internal_type == "BooleanField": converters.append(self.convert_booleanfield_value) elif internal_type == "DateTimeField": if settings.USE_TZ: converters.append(self.convert_datetimefield_value) elif internal_type == "UUIDField": converters.append(self.convert_uuidfield_value) return converters def convert_booleanfield_value(self, value, expression, connection): if value in (0, 1): value = bool(value) return value def convert_datetimefield_value(self, value, expression, connection): if value is not None: value = timezone.make_aware(value, self.connection.timezone) return value def convert_uuidfield_value(self, value, expression, connection): if value is not None: value = uuid.UUID(value) return value def binary_placeholder_sql(self, value): return ( "_binary %s" if value is not None and not hasattr(value, "as_sql") else "%s" ) def subtract_temporals(self, internal_type, lhs, rhs): lhs_sql, lhs_params = lhs rhs_sql, rhs_params = rhs if internal_type == "TimeField": if self.connection.mysql_is_mariadb: # MariaDB includes the microsecond component in TIME_TO_SEC as # a decimal. MySQL returns an integer without microseconds. return ( "CAST((TIME_TO_SEC(%(lhs)s) - TIME_TO_SEC(%(rhs)s)) " "* 1000000 AS SIGNED)" ) % { "lhs": lhs_sql, "rhs": rhs_sql, }, ( *lhs_params, *rhs_params, ) return ( "((TIME_TO_SEC(%(lhs)s) * 1000000 + MICROSECOND(%(lhs)s)) -" " (TIME_TO_SEC(%(rhs)s) * 1000000 + MICROSECOND(%(rhs)s)))" ) % {"lhs": lhs_sql, "rhs": rhs_sql}, tuple(lhs_params) * 2 + tuple( rhs_params ) * 2 params = (*rhs_params, *lhs_params) return "TIMESTAMPDIFF(MICROSECOND, %s, %s)" % (rhs_sql, lhs_sql), params def explain_query_prefix(self, format=None, **options): # Alias MySQL's TRADITIONAL to TEXT for consistency with other backends. if format and format.upper() == "TEXT": format = "TRADITIONAL" elif ( not format and "TREE" in self.connection.features.supported_explain_formats ): # Use TREE by default (if supported) as it's more informative. format = "TREE" analyze = options.pop("analyze", False) prefix = super().explain_query_prefix(format, **options) if analyze and self.connection.features.supports_explain_analyze: # MariaDB uses ANALYZE instead of EXPLAIN ANALYZE. prefix = ( "ANALYZE" if self.connection.mysql_is_mariadb else prefix + " ANALYZE" ) if format and not (analyze and not self.connection.mysql_is_mariadb): # Only MariaDB supports the analyze option with formats. prefix += " FORMAT=%s" % format return prefix def regex_lookup(self, lookup_type): # REGEXP_LIKE doesn't exist in MariaDB. if self.connection.mysql_is_mariadb: if lookup_type == "regex": return "%s REGEXP BINARY %s" return "%s REGEXP %s" match_option = "c" if lookup_type == "regex" else "i" return "REGEXP_LIKE(%%s, %%s, '%s')" % match_option def insert_statement(self, on_conflict=None): if on_conflict == OnConflict.IGNORE: return "INSERT IGNORE INTO" return super().insert_statement(on_conflict=on_conflict) def lookup_cast(self, lookup_type, internal_type=None): lookup = "%s" if internal_type == "JSONField": if self.connection.mysql_is_mariadb or lookup_type in ( "iexact", "contains", "icontains", "startswith", "istartswith", "endswith", "iendswith", "regex", "iregex", ): lookup = "JSON_UNQUOTE(%s)" return lookup def conditional_expression_supported_in_where_clause(self, expression): # MySQL ignores indexes with boolean fields unless they're compared # directly to a boolean value. if isinstance(expression, (Exists, Lookup)): return True if isinstance(expression, ExpressionWrapper) and expression.conditional: return self.conditional_expression_supported_in_where_clause( expression.expression ) if getattr(expression, "conditional", False): return False return super().conditional_expression_supported_in_where_clause(expression) def on_conflict_suffix_sql(self, fields, on_conflict, update_fields, unique_fields): if on_conflict == OnConflict.UPDATE: conflict_suffix_sql = "ON DUPLICATE KEY UPDATE %(fields)s" # The use of VALUES() is deprecated in MySQL 8.0.20+. Instead, use # aliases for the new row and its columns available in MySQL # 8.0.19+. if not self.connection.mysql_is_mariadb: if self.connection.mysql_version >= (8, 0, 19): conflict_suffix_sql = f"AS new {conflict_suffix_sql}" field_sql = "%(field)s = new.%(field)s" else: field_sql = "%(field)s = VALUES(%(field)s)" # Use VALUE() on MariaDB. else: field_sql = "%(field)s = VALUE(%(field)s)" fields = ", ".join( [ field_sql % {"field": field} for field in map(self.quote_name, update_fields) ] ) return conflict_suffix_sql % {"fields": fields} return super().on_conflict_suffix_sql( fields, on_conflict, update_fields, unique_fields, )