naming.py 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210
  1. # sqlalchemy/naming.py
  2. # Copyright (C) 2005-2022 the SQLAlchemy authors and contributors
  3. # <see AUTHORS file>
  4. #
  5. # This module is part of SQLAlchemy and is released under
  6. # the MIT License: https://www.opensource.org/licenses/mit-license.php
  7. """Establish constraint and index naming conventions.
  8. """
  9. import re
  10. from . import events # noqa
  11. from .elements import _NONE_NAME
  12. from .elements import conv
  13. from .schema import CheckConstraint
  14. from .schema import Column
  15. from .schema import Constraint
  16. from .schema import ForeignKeyConstraint
  17. from .schema import Index
  18. from .schema import PrimaryKeyConstraint
  19. from .schema import Table
  20. from .schema import UniqueConstraint
  21. from .. import event
  22. from .. import exc
  23. class ConventionDict(object):
  24. def __init__(self, const, table, convention):
  25. self.const = const
  26. self._is_fk = isinstance(const, ForeignKeyConstraint)
  27. self.table = table
  28. self.convention = convention
  29. self._const_name = const.name
  30. def _key_table_name(self):
  31. return self.table.name
  32. def _column_X(self, idx, attrname):
  33. if self._is_fk:
  34. try:
  35. fk = self.const.elements[idx]
  36. except IndexError:
  37. return ""
  38. else:
  39. return getattr(fk.parent, attrname)
  40. else:
  41. cols = list(self.const.columns)
  42. try:
  43. col = cols[idx]
  44. except IndexError:
  45. return ""
  46. else:
  47. return getattr(col, attrname)
  48. def _key_constraint_name(self):
  49. if self._const_name in (None, _NONE_NAME):
  50. raise exc.InvalidRequestError(
  51. "Naming convention including "
  52. "%(constraint_name)s token requires that "
  53. "constraint is explicitly named."
  54. )
  55. if not isinstance(self._const_name, conv):
  56. self.const.name = None
  57. return self._const_name
  58. def _key_column_X_key(self, idx):
  59. # note this method was missing before
  60. # [ticket:3989], meaning tokens like ``%(column_0_key)s`` weren't
  61. # working even though documented.
  62. return self._column_X(idx, "key")
  63. def _key_column_X_name(self, idx):
  64. return self._column_X(idx, "name")
  65. def _key_column_X_label(self, idx):
  66. return self._column_X(idx, "_ddl_label")
  67. def _key_referred_table_name(self):
  68. fk = self.const.elements[0]
  69. refs = fk.target_fullname.split(".")
  70. if len(refs) == 3:
  71. refschema, reftable, refcol = refs
  72. else:
  73. reftable, refcol = refs
  74. return reftable
  75. def _key_referred_column_X_name(self, idx):
  76. fk = self.const.elements[idx]
  77. # note that before [ticket:3989], this method was returning
  78. # the specification for the :class:`.ForeignKey` itself, which normally
  79. # would be using the ``.key`` of the column, not the name.
  80. return fk.column.name
  81. def __getitem__(self, key):
  82. if key in self.convention:
  83. return self.convention[key](self.const, self.table)
  84. elif hasattr(self, "_key_%s" % key):
  85. return getattr(self, "_key_%s" % key)()
  86. else:
  87. col_template = re.match(r".*_?column_(\d+)(_?N)?_.+", key)
  88. if col_template:
  89. idx = col_template.group(1)
  90. multiples = col_template.group(2)
  91. if multiples:
  92. if self._is_fk:
  93. elems = self.const.elements
  94. else:
  95. elems = list(self.const.columns)
  96. tokens = []
  97. for idx, elem in enumerate(elems):
  98. attr = "_key_" + key.replace("0" + multiples, "X")
  99. try:
  100. tokens.append(getattr(self, attr)(idx))
  101. except AttributeError:
  102. raise KeyError(key)
  103. sep = "_" if multiples.startswith("_") else ""
  104. return sep.join(tokens)
  105. else:
  106. attr = "_key_" + key.replace(idx, "X")
  107. idx = int(idx)
  108. if hasattr(self, attr):
  109. return getattr(self, attr)(idx)
  110. raise KeyError(key)
  111. _prefix_dict = {
  112. Index: "ix",
  113. PrimaryKeyConstraint: "pk",
  114. CheckConstraint: "ck",
  115. UniqueConstraint: "uq",
  116. ForeignKeyConstraint: "fk",
  117. }
  118. def _get_convention(dict_, key):
  119. for super_ in key.__mro__:
  120. if super_ in _prefix_dict and _prefix_dict[super_] in dict_:
  121. return dict_[_prefix_dict[super_]]
  122. elif super_ in dict_:
  123. return dict_[super_]
  124. else:
  125. return None
  126. def _constraint_name_for_table(const, table):
  127. metadata = table.metadata
  128. convention = _get_convention(metadata.naming_convention, type(const))
  129. if isinstance(const.name, conv):
  130. return const.name
  131. elif (
  132. convention is not None
  133. and not isinstance(const.name, conv)
  134. and (
  135. const.name is None
  136. or "constraint_name" in convention
  137. or const.name is _NONE_NAME
  138. )
  139. ):
  140. return conv(
  141. convention
  142. % ConventionDict(const, table, metadata.naming_convention)
  143. )
  144. elif convention is _NONE_NAME:
  145. return None
  146. @event.listens_for(
  147. PrimaryKeyConstraint, "_sa_event_column_added_to_pk_constraint"
  148. )
  149. def _column_added_to_pk_constraint(pk_constraint, col):
  150. if pk_constraint._implicit_generated:
  151. # only operate upon the "implicit" pk constraint for now,
  152. # as we have to force the name to None to reset it. the
  153. # "implicit" constraint will only have a naming convention name
  154. # if at all.
  155. table = pk_constraint.table
  156. pk_constraint.name = None
  157. newname = _constraint_name_for_table(pk_constraint, table)
  158. if newname:
  159. pk_constraint.name = newname
  160. @event.listens_for(Constraint, "after_parent_attach")
  161. @event.listens_for(Index, "after_parent_attach")
  162. def _constraint_name(const, table):
  163. if isinstance(table, Column):
  164. # this path occurs for a CheckConstraint linked to a Column
  165. # for column-attached constraint, set another event
  166. # to link the column attached to the table as this constraint
  167. # associated with the table.
  168. event.listen(
  169. table,
  170. "after_parent_attach",
  171. lambda col, table: _constraint_name(const, table),
  172. )
  173. elif isinstance(table, Table):
  174. if isinstance(const.name, conv) or const.name is _NONE_NAME:
  175. return
  176. newname = _constraint_name_for_table(const, table)
  177. if newname:
  178. const.name = newname