我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用sqlalchemy.orm.session.set()。
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)
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