MCPcopy Create free account
hub / github.com/sqlalchemy/sqlalchemy / CascadeOptions

Class CascadeOptions

lib/sqlalchemy/orm/util.py:180–250  ·  view source on GitHub ↗

Keeps track of the options sent to :paramref:`.relationship.cascade`

Source from the content-addressed store, hash-verified

178
179
180class CascadeOptions(FrozenSet[str]):
181 """Keeps track of the options sent to
182 :paramref:`.relationship.cascade`"""
183
184 _add_w_all_cascades = all_cascades.difference(
185 ["all", "none", "delete-orphan"]
186 )
187 _allowed_cascades = all_cascades
188
189 _viewonly_cascades = ["expunge", "all", "none", "refresh-expire", "merge"]
190
191 __slots__ = (
192 "save_update",
193 "delete",
194 "refresh_expire",
195 "merge",
196 "expunge",
197 "delete_orphan",
198 )
199
200 save_update: bool
201 delete: bool
202 refresh_expire: bool
203 merge: bool
204 expunge: bool
205 delete_orphan: bool
206
207 def __new__(
208 cls, value_list: Optional[Union[Iterable[str], str]]
209 ) -> CascadeOptions:
210 if isinstance(value_list, str) or value_list is None:
211 return cls.from_string(value_list) # type: ignore
212 values = set(value_list)
213 if values.difference(cls._allowed_cascades):
214 raise sa_exc.ArgumentError(
215 "Invalid cascade option(s): %s"
216 % ", ".join(
217 [
218 repr(x)
219 for x in sorted(
220 values.difference(cls._allowed_cascades)
221 )
222 ]
223 )
224 )
225
226 if "all" in values:
227 values.update(cls._add_w_all_cascades)
228 if "none" in values:
229 values.clear()
230 values.discard("all")
231
232 self = super().__new__(cls, values)
233 self.save_update = "save-update" in values
234 self.delete = "delete" in values
235 self.refresh_expire = "refresh-expire" in values
236 self.merge = "merge" in values
237 self.expunge = "expunge" in values

Callers 2

test_write_cascadesMethod · 0.90
_set_cascadeMethod · 0.85

Calls 1

differenceMethod · 0.45

Tested by 1

test_write_cascadesMethod · 0.72