schema.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. import copy
  2. import datetime
  3. import re
  4. from django.db import DatabaseError
  5. from django.db.backends.base.schema import (
  6. BaseDatabaseSchemaEditor,
  7. _related_non_m2m_objects,
  8. )
  9. from django.utils.duration import duration_iso_string
  10. class DatabaseSchemaEditor(BaseDatabaseSchemaEditor):
  11. sql_create_column = "ALTER TABLE %(table)s ADD %(column)s %(definition)s"
  12. sql_alter_column_type = "MODIFY %(column)s %(type)s%(collation)s"
  13. sql_alter_column_null = "MODIFY %(column)s NULL"
  14. sql_alter_column_not_null = "MODIFY %(column)s NOT NULL"
  15. sql_alter_column_default = "MODIFY %(column)s DEFAULT %(default)s"
  16. sql_alter_column_no_default = "MODIFY %(column)s DEFAULT NULL"
  17. sql_alter_column_no_default_null = sql_alter_column_no_default
  18. sql_delete_column = "ALTER TABLE %(table)s DROP COLUMN %(column)s"
  19. sql_create_column_inline_fk = (
  20. "CONSTRAINT %(name)s REFERENCES %(to_table)s(%(to_column)s)%(deferrable)s"
  21. )
  22. sql_delete_table = "DROP TABLE %(table)s CASCADE CONSTRAINTS"
  23. sql_create_index = "CREATE INDEX %(name)s ON %(table)s (%(columns)s)%(extra)s"
  24. def quote_value(self, value):
  25. if isinstance(value, (datetime.date, datetime.time, datetime.datetime)):
  26. return "'%s'" % value
  27. elif isinstance(value, datetime.timedelta):
  28. return "'%s'" % duration_iso_string(value)
  29. elif isinstance(value, str):
  30. return "'%s'" % value.replace("'", "''")
  31. elif isinstance(value, (bytes, bytearray, memoryview)):
  32. return "'%s'" % value.hex()
  33. elif isinstance(value, bool):
  34. return "1" if value else "0"
  35. else:
  36. return str(value)
  37. def remove_field(self, model, field):
  38. # If the column is an identity column, drop the identity before
  39. # removing the field.
  40. if self._is_identity_column(model._meta.db_table, field.column):
  41. self._drop_identity(model._meta.db_table, field.column)
  42. super().remove_field(model, field)
  43. def delete_model(self, model):
  44. # Run superclass action
  45. super().delete_model(model)
  46. # Clean up manually created sequence.
  47. self.execute(
  48. """
  49. DECLARE
  50. i INTEGER;
  51. BEGIN
  52. SELECT COUNT(1) INTO i FROM USER_SEQUENCES
  53. WHERE SEQUENCE_NAME = '%(sq_name)s';
  54. IF i = 1 THEN
  55. EXECUTE IMMEDIATE 'DROP SEQUENCE "%(sq_name)s"';
  56. END IF;
  57. END;
  58. /"""
  59. % {
  60. "sq_name": self.connection.ops._get_no_autofield_sequence_name(
  61. model._meta.db_table
  62. )
  63. }
  64. )
  65. def alter_field(self, model, old_field, new_field, strict=False):
  66. try:
  67. super().alter_field(model, old_field, new_field, strict)
  68. except DatabaseError as e:
  69. description = str(e)
  70. # If we're changing type to an unsupported type we need a
  71. # SQLite-ish workaround
  72. if "ORA-22858" in description or "ORA-22859" in description:
  73. self._alter_field_type_workaround(model, old_field, new_field)
  74. # If an identity column is changing to a non-numeric type, drop the
  75. # identity first.
  76. elif "ORA-30675" in description:
  77. self._drop_identity(model._meta.db_table, old_field.column)
  78. self.alter_field(model, old_field, new_field, strict)
  79. # If a primary key column is changing to an identity column, drop
  80. # the primary key first.
  81. elif "ORA-30673" in description and old_field.primary_key:
  82. self._delete_primary_key(model, strict=True)
  83. self._alter_field_type_workaround(model, old_field, new_field)
  84. # If a collation is changing on a primary key, drop the primary key
  85. # first.
  86. elif "ORA-43923" in description and old_field.primary_key:
  87. self._delete_primary_key(model, strict=True)
  88. self.alter_field(model, old_field, new_field, strict)
  89. # Restore a primary key, if needed.
  90. if new_field.primary_key:
  91. self.execute(self._create_primary_key_sql(model, new_field))
  92. else:
  93. raise
  94. def _alter_field_type_workaround(self, model, old_field, new_field):
  95. """
  96. Oracle refuses to change from some type to other type.
  97. What we need to do instead is:
  98. - Add a nullable version of the desired field with a temporary name. If
  99. the new column is an auto field, then the temporary column can't be
  100. nullable.
  101. - Update the table to transfer values from old to new
  102. - Drop old column
  103. - Rename the new column and possibly drop the nullable property
  104. """
  105. # Make a new field that's like the new one but with a temporary
  106. # column name.
  107. new_temp_field = copy.deepcopy(new_field)
  108. new_temp_field.null = new_field.get_internal_type() not in (
  109. "AutoField",
  110. "BigAutoField",
  111. "SmallAutoField",
  112. )
  113. new_temp_field.column = self._generate_temp_name(new_field.column)
  114. # Add it
  115. self.add_field(model, new_temp_field)
  116. # Explicit data type conversion
  117. # https://docs.oracle.com/en/database/oracle/oracle-database/21/sqlrf
  118. # /Data-Type-Comparison-Rules.html#GUID-D0C5A47E-6F93-4C2D-9E49-4F2B86B359DD
  119. new_value = self.quote_name(old_field.column)
  120. old_type = old_field.db_type(self.connection)
  121. if re.match("^N?CLOB", old_type):
  122. new_value = "TO_CHAR(%s)" % new_value
  123. old_type = "VARCHAR2"
  124. if re.match("^N?VARCHAR2", old_type):
  125. new_internal_type = new_field.get_internal_type()
  126. if new_internal_type == "DateField":
  127. new_value = "TO_DATE(%s, 'YYYY-MM-DD')" % new_value
  128. elif new_internal_type == "DateTimeField":
  129. new_value = "TO_TIMESTAMP(%s, 'YYYY-MM-DD HH24:MI:SS.FF')" % new_value
  130. elif new_internal_type == "TimeField":
  131. # TimeField are stored as TIMESTAMP with a 1900-01-01 date part.
  132. new_value = "CONCAT('1900-01-01 ', %s)" % new_value
  133. new_value = "TO_TIMESTAMP(%s, 'YYYY-MM-DD HH24:MI:SS.FF')" % new_value
  134. # Transfer values across
  135. self.execute(
  136. "UPDATE %s set %s=%s"
  137. % (
  138. self.quote_name(model._meta.db_table),
  139. self.quote_name(new_temp_field.column),
  140. new_value,
  141. )
  142. )
  143. # Drop the old field
  144. self.remove_field(model, old_field)
  145. # Rename and possibly make the new field NOT NULL
  146. super().alter_field(model, new_temp_field, new_field)
  147. # Recreate foreign key (if necessary) because the old field is not
  148. # passed to the alter_field() and data types of new_temp_field and
  149. # new_field always match.
  150. new_type = new_field.db_type(self.connection)
  151. if (
  152. (old_field.primary_key and new_field.primary_key)
  153. or (old_field.unique and new_field.unique)
  154. ) and old_type != new_type:
  155. for _, rel in _related_non_m2m_objects(new_temp_field, new_field):
  156. if rel.field.db_constraint:
  157. self.execute(
  158. self._create_fk_sql(rel.related_model, rel.field, "_fk")
  159. )
  160. def _alter_column_type_sql(
  161. self, model, old_field, new_field, new_type, old_collation, new_collation
  162. ):
  163. auto_field_types = {"AutoField", "BigAutoField", "SmallAutoField"}
  164. # Drop the identity if migrating away from AutoField.
  165. if (
  166. old_field.get_internal_type() in auto_field_types
  167. and new_field.get_internal_type() not in auto_field_types
  168. and self._is_identity_column(model._meta.db_table, new_field.column)
  169. ):
  170. self._drop_identity(model._meta.db_table, new_field.column)
  171. return super()._alter_column_type_sql(
  172. model, old_field, new_field, new_type, old_collation, new_collation
  173. )
  174. def normalize_name(self, name):
  175. """
  176. Get the properly shortened and uppercased identifier as returned by
  177. quote_name() but without the quotes.
  178. """
  179. nn = self.quote_name(name)
  180. if nn[0] == '"' and nn[-1] == '"':
  181. nn = nn[1:-1]
  182. return nn
  183. def _generate_temp_name(self, for_name):
  184. """Generate temporary names for workarounds that need temp columns."""
  185. suffix = hex(hash(for_name)).upper()[1:]
  186. return self.normalize_name(for_name + "_" + suffix)
  187. def prepare_default(self, value):
  188. return self.quote_value(value)
  189. def _field_should_be_indexed(self, model, field):
  190. create_index = super()._field_should_be_indexed(model, field)
  191. db_type = field.db_type(self.connection)
  192. if (
  193. db_type is not None
  194. and db_type.lower() in self.connection._limited_data_types
  195. ):
  196. return False
  197. return create_index
  198. def _is_identity_column(self, table_name, column_name):
  199. with self.connection.cursor() as cursor:
  200. cursor.execute(
  201. """
  202. SELECT
  203. CASE WHEN identity_column = 'YES' THEN 1 ELSE 0 END
  204. FROM user_tab_cols
  205. WHERE table_name = %s AND
  206. column_name = %s
  207. """,
  208. [self.normalize_name(table_name), self.normalize_name(column_name)],
  209. )
  210. row = cursor.fetchone()
  211. return row[0] if row else False
  212. def _drop_identity(self, table_name, column_name):
  213. self.execute(
  214. "ALTER TABLE %(table)s MODIFY %(column)s DROP IDENTITY"
  215. % {
  216. "table": self.quote_name(table_name),
  217. "column": self.quote_name(column_name),
  218. }
  219. )
  220. def _get_default_collation(self, table_name):
  221. with self.connection.cursor() as cursor:
  222. cursor.execute(
  223. """
  224. SELECT default_collation FROM user_tables WHERE table_name = %s
  225. """,
  226. [self.normalize_name(table_name)],
  227. )
  228. return cursor.fetchone()[0]
  229. def _collate_sql(self, collation, old_collation=None, table_name=None):
  230. if collation is None and old_collation is not None:
  231. collation = self._get_default_collation(table_name)
  232. return super()._collate_sql(collation, old_collation, table_name)