Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __get__(self, instance, cls=None):
if instance is None: return self
st = state(instance)
if not st.options.instrument:
return st.document[self.name]
try:
return st.instrumented(self.name)
except KeyError:
value = self.field.schema.validate(S.Missing)
if value is S.Missing:
raise AttributeError(self.name)
else:
st.document[self.name] = value
return value
def _next_impl(self):
doc = next(self.ming_cursor)
obj = self.session.imap.get(self.cls, doc['_id'])
if obj is None:
obj = self.mapper.create(doc, self._options, remake=False)
state(obj).status = ObjectState.clean
self.session.save(obj)
elif self._options.refresh:
# Refresh object
state(obj).update(doc)
state(obj).status = ObjectState.clean
else:
# Never refresh objects from the DB unless explicitly requested
pass
other_session = session(obj)
if other_session is not None and other_session != self:
other_session.expunge(obj)
self.session.save(obj)
if self._options.decorate is not None:
return self._options.decorate(obj)
else:
return obj
If the session has ``autoflush`` option, the session
if flushed before performing the query.
It returns an :class:`.ODMCursor` with the results.
"""
decorate = kwargs.pop('decorate', None)
if self.autoflush:
self.flush()
m = mapper(cls)
obj = self.impl.find_and_modify(m.collection, *args, **kwargs)
if obj is None: return None
cursor = ODMCursor(self, cls, iter([ obj ]), refresh=True, decorate=decorate)
result = cursor.first()
state(result).status = ObjectState.clean
return result
def dirty(self):
return (obj for obj in six.itervalues(self._objects)
if state(obj).status == ObjectState.dirty)
def delete(self):
"""Mark the object for deletion on next flush"""
st = state(self.instance)
st.status = st.deleted
def expunge(self, obj):
"""Remove an object from the Session (and its UnitOfWork and IdentityMap)"""
self.uow.expunge(obj)
self.imap.expunge(obj)
state(obj).session = None
def _next_impl(self):
doc = next(self.ming_cursor)
obj = self.session.imap.get(self.cls, doc['_id'])
if obj is None:
obj = self.mapper.create(doc, self._options, remake=False)
state(obj).status = ObjectState.clean
self.session.save(obj)
elif self._options.refresh:
# Refresh object
state(obj).update(doc)
state(obj).status = ObjectState.clean
else:
# Never refresh objects from the DB unless explicitly requested
pass
other_session = session(obj)
if other_session is not None and other_session != self:
other_session.expunge(obj)
self.session.save(obj)
if self._options.decorate is not None:
return self._options.decorate(obj)
else:
return obj
def get_depot_history(cls, instance):
istate = state(instance)
if not hasattr(istate, '_depot_history'):
istate._depot_history = _DepotHistory()
return istate._depot_history
def _next_impl(self):
doc = next(self.ming_cursor)
obj = self.session.imap.get(self.cls, doc['_id'])
if obj is None:
obj = self.mapper.create(doc, self._options, remake=False)
state(obj).status = ObjectState.clean
self.session.save(obj)
elif self._options.refresh:
# Refresh object
state(obj).update(doc)
state(obj).status = ObjectState.clean
else:
# Never refresh objects from the DB unless explicitly requested
pass
other_session = session(obj)
if other_session is not None and other_session != self:
other_session.expunge(obj)
self.session.save(obj)
if self._options.decorate is not None:
return self._options.decorate(obj)
else:
return obj
def clean(self):
return (obj for obj in six.itervalues(self._objects)
if state(obj).status == ObjectState.clean)