schema.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281
  1. from django.db.backends.base.schema import BaseDatabaseSchemaEditor
  2. from django.db.models import NOT_PROVIDED, F, UniqueConstraint
  3. from django.db.models.constants import LOOKUP_SEP
  4. class DatabaseSchemaEditor(BaseDatabaseSchemaEditor):
  5. sql_rename_table = "RENAME TABLE %(old_table)s TO %(new_table)s"
  6. sql_alter_column_null = "MODIFY %(column)s %(type)s NULL"
  7. sql_alter_column_not_null = "MODIFY %(column)s %(type)s NOT NULL"
  8. sql_alter_column_type = "MODIFY %(column)s %(type)s%(collation)s%(comment)s"
  9. sql_alter_column_no_default_null = "ALTER COLUMN %(column)s SET DEFAULT NULL"
  10. # No 'CASCADE' which works as a no-op in MySQL but is undocumented
  11. sql_delete_column = "ALTER TABLE %(table)s DROP COLUMN %(column)s"
  12. sql_delete_unique = "ALTER TABLE %(table)s DROP INDEX %(name)s"
  13. sql_create_column_inline_fk = (
  14. ", ADD CONSTRAINT %(name)s FOREIGN KEY (%(column)s) "
  15. "REFERENCES %(to_table)s(%(to_column)s)"
  16. )
  17. sql_delete_fk = "ALTER TABLE %(table)s DROP FOREIGN KEY %(name)s"
  18. sql_delete_index = "DROP INDEX %(name)s ON %(table)s"
  19. sql_rename_index = "ALTER TABLE %(table)s RENAME INDEX %(old_name)s TO %(new_name)s"
  20. sql_create_pk = (
  21. "ALTER TABLE %(table)s ADD CONSTRAINT %(name)s PRIMARY KEY (%(columns)s)"
  22. )
  23. sql_delete_pk = "ALTER TABLE %(table)s DROP PRIMARY KEY"
  24. sql_create_index = "CREATE INDEX %(name)s ON %(table)s (%(columns)s)%(extra)s"
  25. sql_alter_table_comment = "ALTER TABLE %(table)s COMMENT = %(comment)s"
  26. sql_alter_column_comment = None
  27. @property
  28. def sql_delete_check(self):
  29. if self.connection.mysql_is_mariadb:
  30. # The name of the column check constraint is the same as the field
  31. # name on MariaDB. Adding IF EXISTS clause prevents migrations
  32. # crash. Constraint is removed during a "MODIFY" column statement.
  33. return "ALTER TABLE %(table)s DROP CONSTRAINT IF EXISTS %(name)s"
  34. return "ALTER TABLE %(table)s DROP CHECK %(name)s"
  35. @property
  36. def sql_rename_column(self):
  37. is_mariadb = self.connection.mysql_is_mariadb
  38. if is_mariadb and self.connection.mysql_version < (10, 5, 2):
  39. # MariaDB < 10.5.2 doesn't support an
  40. # "ALTER TABLE ... RENAME COLUMN" statement.
  41. return "ALTER TABLE %(table)s CHANGE %(old_column)s %(new_column)s %(type)s"
  42. return super().sql_rename_column
  43. def quote_value(self, value):
  44. self.connection.ensure_connection()
  45. # MySQLdb escapes to string, PyMySQL to bytes.
  46. quoted = self.connection.connection.escape(
  47. value, self.connection.connection.encoders
  48. )
  49. if isinstance(value, str) and isinstance(quoted, bytes):
  50. quoted = quoted.decode()
  51. return quoted
  52. def _is_limited_data_type(self, field):
  53. db_type = field.db_type(self.connection)
  54. return (
  55. db_type is not None
  56. and db_type.lower() in self.connection._limited_data_types
  57. )
  58. def _is_text_or_blob(self, field):
  59. db_type = field.db_type(self.connection)
  60. return db_type and db_type.lower().endswith(("blob", "text"))
  61. def skip_default(self, field):
  62. default_is_empty = self.effective_default(field) in ("", b"")
  63. if default_is_empty and self._is_text_or_blob(field):
  64. return True
  65. if not self._supports_limited_data_type_defaults:
  66. return self._is_limited_data_type(field)
  67. return False
  68. def skip_default_on_alter(self, field):
  69. default_is_empty = self.effective_default(field) in ("", b"")
  70. if default_is_empty and self._is_text_or_blob(field):
  71. return True
  72. if self._is_limited_data_type(field) and not self.connection.mysql_is_mariadb:
  73. # MySQL doesn't support defaults for BLOB and TEXT in the
  74. # ALTER COLUMN statement.
  75. return True
  76. return False
  77. @property
  78. def _supports_limited_data_type_defaults(self):
  79. # MariaDB and MySQL >= 8.0.13 support defaults for BLOB and TEXT.
  80. if self.connection.mysql_is_mariadb:
  81. return True
  82. return self.connection.mysql_version >= (8, 0, 13)
  83. def _column_default_sql(self, field):
  84. if (
  85. not self.connection.mysql_is_mariadb
  86. and self._supports_limited_data_type_defaults
  87. and self._is_limited_data_type(field)
  88. ):
  89. # MySQL supports defaults for BLOB and TEXT columns only if the
  90. # default value is written as an expression i.e. in parentheses.
  91. return "(%s)"
  92. return super()._column_default_sql(field)
  93. def add_field(self, model, field):
  94. super().add_field(model, field)
  95. # Simulate the effect of a one-off default.
  96. # field.default may be unhashable, so a set isn't used for "in" check.
  97. if self.skip_default(field) and field.default not in (None, NOT_PROVIDED):
  98. effective_default = self.effective_default(field)
  99. self.execute(
  100. "UPDATE %(table)s SET %(column)s = %%s"
  101. % {
  102. "table": self.quote_name(model._meta.db_table),
  103. "column": self.quote_name(field.column),
  104. },
  105. [effective_default],
  106. )
  107. def remove_constraint(self, model, constraint):
  108. if (
  109. isinstance(constraint, UniqueConstraint)
  110. and constraint.create_sql(model, self) is not None
  111. ):
  112. self._create_missing_fk_index(
  113. model,
  114. fields=constraint.fields,
  115. expressions=constraint.expressions,
  116. )
  117. super().remove_constraint(model, constraint)
  118. def remove_index(self, model, index):
  119. self._create_missing_fk_index(
  120. model,
  121. fields=[field_name for field_name, _ in index.fields_orders],
  122. expressions=index.expressions,
  123. )
  124. super().remove_index(model, index)
  125. def _field_should_be_indexed(self, model, field):
  126. if not super()._field_should_be_indexed(model, field):
  127. return False
  128. storage = self.connection.introspection.get_storage_engine(
  129. self.connection.cursor(), model._meta.db_table
  130. )
  131. # No need to create an index for ForeignKey fields except if
  132. # db_constraint=False because the index from that constraint won't be
  133. # created.
  134. if (
  135. storage == "InnoDB"
  136. and field.get_internal_type() == "ForeignKey"
  137. and field.db_constraint
  138. ):
  139. return False
  140. return not self._is_limited_data_type(field)
  141. def _create_missing_fk_index(
  142. self,
  143. model,
  144. *,
  145. fields,
  146. expressions=None,
  147. ):
  148. """
  149. MySQL can remove an implicit FK index on a field when that field is
  150. covered by another index like a unique_together. "covered" here means
  151. that the more complex index has the FK field as its first field (see
  152. https://bugs.mysql.com/bug.php?id=37910).
  153. Manually create an implicit FK index to make it possible to remove the
  154. composed index.
  155. """
  156. first_field_name = None
  157. if fields:
  158. first_field_name = fields[0]
  159. elif (
  160. expressions
  161. and self.connection.features.supports_expression_indexes
  162. and isinstance(expressions[0], F)
  163. and LOOKUP_SEP not in expressions[0].name
  164. ):
  165. first_field_name = expressions[0].name
  166. if not first_field_name:
  167. return
  168. first_field = model._meta.get_field(first_field_name)
  169. if first_field.get_internal_type() == "ForeignKey":
  170. column = self.connection.introspection.identifier_converter(
  171. first_field.column
  172. )
  173. with self.connection.cursor() as cursor:
  174. constraint_names = [
  175. name
  176. for name, infodict in self.connection.introspection.get_constraints(
  177. cursor, model._meta.db_table
  178. ).items()
  179. if infodict["index"] and infodict["columns"][0] == column
  180. ]
  181. # There are no other indexes that starts with the FK field, only
  182. # the index that is expected to be deleted.
  183. if len(constraint_names) == 1:
  184. self.execute(
  185. self._create_index_sql(model, fields=[first_field], suffix="")
  186. )
  187. def _delete_composed_index(self, model, fields, *args):
  188. self._create_missing_fk_index(model, fields=fields)
  189. return super()._delete_composed_index(model, fields, *args)
  190. def _set_field_new_type(self, field, new_type):
  191. """
  192. Keep the NULL and DEFAULT properties of the old field. If it has
  193. changed, it will be handled separately.
  194. """
  195. if field.db_default is not NOT_PROVIDED:
  196. default_sql, params = self.db_default_sql(field)
  197. default_sql %= tuple(self.quote_value(p) for p in params)
  198. new_type += f" DEFAULT {default_sql}"
  199. if field.null:
  200. new_type += " NULL"
  201. else:
  202. new_type += " NOT NULL"
  203. return new_type
  204. def _alter_column_type_sql(
  205. self, model, old_field, new_field, new_type, old_collation, new_collation
  206. ):
  207. new_type = self._set_field_new_type(old_field, new_type)
  208. return super()._alter_column_type_sql(
  209. model, old_field, new_field, new_type, old_collation, new_collation
  210. )
  211. def _field_db_check(self, field, field_db_params):
  212. if self.connection.mysql_is_mariadb and self.connection.mysql_version >= (
  213. 10,
  214. 5,
  215. 2,
  216. ):
  217. return super()._field_db_check(field, field_db_params)
  218. # On MySQL and MariaDB < 10.5.2 (no support for
  219. # "ALTER TABLE ... RENAME COLUMN" statements), check constraints with
  220. # the column name as it requires explicit recreation when the column is
  221. # renamed.
  222. return field_db_params["check"]
  223. def _rename_field_sql(self, table, old_field, new_field, new_type):
  224. new_type = self._set_field_new_type(old_field, new_type)
  225. return super()._rename_field_sql(table, old_field, new_field, new_type)
  226. def _alter_column_comment_sql(self, model, new_field, new_type, new_db_comment):
  227. # Comment is alter when altering the column type.
  228. return "", []
  229. def _comment_sql(self, comment):
  230. comment_sql = super()._comment_sql(comment)
  231. return f" COMMENT {comment_sql}"
  232. def _alter_column_null_sql(self, model, old_field, new_field):
  233. if new_field.db_default is NOT_PROVIDED:
  234. return super()._alter_column_null_sql(model, old_field, new_field)
  235. new_db_params = new_field.db_parameters(connection=self.connection)
  236. type_sql = self._set_field_new_type(new_field, new_db_params["type"])
  237. return (
  238. "MODIFY %(column)s %(type)s"
  239. % {
  240. "column": self.quote_name(new_field.column),
  241. "type": type_sql,
  242. },
  243. [],
  244. )