MCPcopy Index your code
hub / github.com/sqlalchemy/sqlalchemy / CascadeOptions

Class CascadeOptions

lib/sqlalchemy/orm/util.py:184–254  ·  view source on GitHub ↗

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

Source from the content-addressed store, hash-verified

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