Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/naming.py: 45%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

120 statements  

1# sql/naming.py 

2# Copyright (C) 2005-2026 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# mypy: allow-untyped-defs, allow-untyped-calls 

8 

9"""Establish constraint and index naming conventions.""" 

10 

11from __future__ import annotations 

12 

13import re 

14 

15from . import events # noqa 

16from .base import _NONE_NAME 

17from .elements import conv as conv 

18from .schema import CheckConstraint 

19from .schema import Column 

20from .schema import Constraint 

21from .schema import ForeignKeyConstraint 

22from .schema import Index 

23from .schema import PrimaryKeyConstraint 

24from .schema import Table 

25from .schema import UniqueConstraint 

26from .. import event 

27from .. import exc 

28 

29 

30class ConventionDict: 

31 def __init__(self, const, table, convention): 

32 self.const = const 

33 self._is_fk = isinstance(const, ForeignKeyConstraint) 

34 self.table = table 

35 self.convention = convention 

36 self._const_name = const.name 

37 

38 def _key_table_name(self): 

39 return self.table.name 

40 

41 def _column_X(self, idx, attrname): 

42 if self._is_fk: 

43 try: 

44 fk = self.const.elements[idx] 

45 except IndexError: 

46 return "" 

47 else: 

48 return getattr(fk.parent, attrname) 

49 else: 

50 cols = list(self.const.columns) 

51 try: 

52 col = cols[idx] 

53 except IndexError: 

54 return "" 

55 else: 

56 return getattr(col, attrname) 

57 

58 def _key_constraint_name(self): 

59 if self._const_name in (None, _NONE_NAME): 

60 raise exc.InvalidRequestError( 

61 "Naming convention including " 

62 "%(constraint_name)s token requires that " 

63 "constraint is explicitly named." 

64 ) 

65 if not isinstance(self._const_name, conv): 

66 self.const.name = None 

67 return self._const_name 

68 

69 def _key_column_X_key(self, idx): 

70 # note this method was missing before 

71 # [ticket:3989], meaning tokens like ``%(column_0_key)s`` weren't 

72 # working even though documented. 

73 return self._column_X(idx, "key") 

74 

75 def _key_column_X_name(self, idx): 

76 return self._column_X(idx, "name") 

77 

78 def _key_column_X_label(self, idx): 

79 return self._column_X(idx, "_ddl_label") 

80 

81 def _key_referred_table_name(self): 

82 fk = self.const.elements[0] 

83 return fk.target_tokens.table_name 

84 

85 def _key_referred_column_X_name(self, idx): 

86 fk = self.const.elements[idx] 

87 # note that before [ticket:3989], this method was returning 

88 # the specification for the :class:`.ForeignKey` itself, which normally 

89 # would be using the ``.key`` of the column, not the name. 

90 return fk.column.name 

91 

92 def __getitem__(self, key): 

93 if key in self.convention: 

94 return self.convention[key](self.const, self.table) 

95 elif hasattr(self, "_key_%s" % key): 

96 return getattr(self, "_key_%s" % key)() 

97 else: 

98 col_template = re.match(r".*_?column_(\d+)(_?N)?_.+", key) 

99 if col_template: 

100 idx = col_template.group(1) 

101 multiples = col_template.group(2) 

102 

103 if multiples: 

104 if self._is_fk: 

105 elems = self.const.elements 

106 else: 

107 elems = list(self.const.columns) 

108 tokens = [] 

109 for idx, elem in enumerate(elems): 

110 attr = "_key_" + key.replace("0" + multiples, "X") 

111 try: 

112 tokens.append(getattr(self, attr)(idx)) 

113 except AttributeError: 

114 raise KeyError(key) 

115 sep = "_" if multiples.startswith("_") else "" 

116 return sep.join(tokens) 

117 else: 

118 attr = "_key_" + key.replace(idx, "X") 

119 idx = int(idx) 

120 if hasattr(self, attr): 

121 return getattr(self, attr)(idx) 

122 raise KeyError(key) 

123 

124 

125_prefix_dict = { 

126 Index: "ix", 

127 PrimaryKeyConstraint: "pk", 

128 CheckConstraint: "ck", 

129 UniqueConstraint: "uq", 

130 ForeignKeyConstraint: "fk", 

131} 

132 

133 

134def _get_convention(dict_, key): 

135 for super_ in key.__mro__: 

136 if super_ in _prefix_dict and _prefix_dict[super_] in dict_: 

137 return dict_[_prefix_dict[super_]] 

138 elif super_ in dict_: 

139 return dict_[super_] 

140 else: 

141 return None 

142 

143 

144def _constraint_name_for_table(const, table): 

145 metadata = table.metadata 

146 convention = _get_convention(metadata.naming_convention, type(const)) 

147 

148 if isinstance(const.name, conv): 

149 return const.name 

150 elif ( 

151 convention is not None 

152 and not isinstance(const.name, conv) 

153 and ( 

154 const.name is None 

155 or "constraint_name" in convention 

156 or const.name is _NONE_NAME 

157 ) 

158 ): 

159 return conv( 

160 convention 

161 % ConventionDict(const, table, metadata.naming_convention) 

162 ) 

163 elif convention is _NONE_NAME: 

164 return None 

165 

166 

167@event.listens_for( 

168 PrimaryKeyConstraint, "_sa_event_column_added_to_pk_constraint" 

169) 

170def _column_added_to_pk_constraint(pk_constraint, col): 

171 if pk_constraint._implicit_generated: 

172 # only operate upon the "implicit" pk constraint for now, 

173 # as we have to force the name to None to reset it. the 

174 # "implicit" constraint will only have a naming convention name 

175 # if at all. 

176 table = pk_constraint.table 

177 pk_constraint.name = None 

178 newname = _constraint_name_for_table(pk_constraint, table) 

179 if newname: 

180 pk_constraint.name = newname 

181 

182 

183@event.listens_for(Constraint, "after_parent_attach") 

184@event.listens_for(Index, "after_parent_attach") 

185def _constraint_name(const, table): 

186 if isinstance(table, Column): 

187 # this path occurs for a CheckConstraint linked to a Column 

188 

189 # for column-attached constraint, set another event 

190 # to link the column attached to the table as this constraint 

191 # associated with the table. 

192 event.listen( 

193 table, 

194 "after_parent_attach", 

195 lambda col, table: _constraint_name(const, table), 

196 ) 

197 

198 elif isinstance(table, Table): 

199 if isinstance(const.name, conv) or const.name is _NONE_NAME: 

200 return 

201 

202 newname = _constraint_name_for_table(const, table) 

203 if newname: 

204 const.name = newname