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