def _validator_events(desc, key, validator, include_removes):
"""Runs a validation method on an attribute value to be set or appended."""
if include_removes:
def append(state, value, initiator):
return validator(state.obj(), key, value, False)
def set_(state, value, oldvalue, initiator):
return validator(state.obj(), key, value, False)
def remove(state, value, initiator):
validator(state.obj(), key, value, True)
else:
def append(state, value, initiator):
return validator(state.obj(), key, value)
def set_(state, value, oldvalue, initiator):
return validator(state.obj(), key, value)
event.listen(desc, 'append', append, raw=True, retval=True)
event.listen(desc, 'set', set_, raw=True, retval=True)
if include_removes:
event.listen(desc, "remove", remove, raw=True, retval=True)
python类set()的实例源码
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, arg):
values = set([
c for c
in re.split('\s*,\s*', arg or "")
if c
])
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(
values.difference(cls._allowed_cascades)
)])
)
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, arg):
values = set([
c for c
in re.split('\s*,\s*', arg or "")
if c
])
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(
values.difference(cls._allowed_cascades)
)])
)
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, arg):
values = set([
c for c
in re.split('\s*,\s*', arg or "")
if c
])
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(
values.difference(cls._allowed_cascades)
)])
)
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def set(self, reg, key, value):
reg._attributes[(key, self.path)] = value
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the fundamental
mapping and operation that's failing, while still reproducing the issue
on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
def randomize_unitofwork():
"""Use random-ordering sets within the unit of work in order
to detect unit of work sorting issues.
This is a utility function that can be used to help reproduce
inconsistent unit of work sorting issues. For example,
if two kinds of objects A and B are being inserted, and
B has a foreign key reference to A - the A must be inserted first.
However, if there is no relationship between A and B, the unit of work
won't know to perform this sorting, and an operation may or may not
fail, depending on how the ordering works out. Since Python sets
and dictionaries have non-deterministic ordering, such an issue may
occur on some runs and not on others, and in practice it tends to
have a great dependence on the state of the interpreter. This leads
to so-called "heisenbugs" where changing entirely irrelevant aspects
of the test program still cause the failure behavior to change.
By calling ``randomize_unitofwork()`` when a script first runs, the
ordering of a key series of sets within the unit of work implementation
are randomized, so that the script can be minimized down to the
fundamental mapping and operation that's failing, while still reproducing
the issue on at least some runs.
This utility is also available when running the test suite via the
``--reversetop`` flag.
.. versionadded:: 0.8.1 created a standalone version of the
``--reversetop`` feature.
"""
from sqlalchemy.orm import unitofwork, session, mapper, dependency
from sqlalchemy.util import topological
from sqlalchemy.testing.util import RandomSet
topological.set = unitofwork.set = session.set = mapper.set = \
dependency.set = RandomSet
def __new__(cls, value_list):
if isinstance(value_list, util.string_types) or value_list is None:
return cls.from_string(value_list)
values = set(value_list)
if values.difference(cls._allowed_cascades):
raise sa_exc.ArgumentError(
"Invalid cascade option(s): %s" %
", ".join([repr(x) for x in
sorted(values.difference(cls._allowed_cascades))]))
if "all" in values:
values.update(cls._add_w_all_cascades)
if "none" in values:
values.clear()
values.discard('all')
self = frozenset.__new__(CascadeOptions, values)
self.save_update = 'save-update' in values
self.delete = 'delete' in values
self.refresh_expire = 'refresh-expire' in values
self.merge = 'merge' in values
self.expunge = 'expunge' in values
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
util.warn("The 'delete-orphan' cascade "
"option requires 'delete'.")
return self
