mirror of
https://github.com/internetarchive/warcprox.git
synced 2025-01-18 13:22:09 +01:00
Merge pull request #14 from internetarchive/orm-dumb-save
Orm dumb save
This commit is contained in:
Noah Levitt
GitHub
commit
97dca78497
@ -1,7 +1,7 @@
|
||||
'''
|
||||
doublethink/orm.py - rethinkdb ORM
|
||||
|
||||
Copyright (C) 2017 Internet Archive
|
||||
Copyright (C) 2017-2019 Internet Archive
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
@ -20,124 +20,20 @@ import rethinkdb as r
|
||||
import logging
|
||||
import doublethink
|
||||
|
||||
class WatchedDict(dict, object):
|
||||
def __init__(self, d, callback, field):
|
||||
self.callback = callback
|
||||
self.field = field
|
||||
for key in d:
|
||||
dict.__setitem__(self, key, watch(
|
||||
d[key], callback=self.callback, field=self.field))
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
self.callback(self.field)
|
||||
return dict.__setitem__(self, key, watch(
|
||||
value, callback=self.callback, field=self.field))
|
||||
|
||||
def __delitem__(self, key):
|
||||
self.callback(self.field)
|
||||
return dict.__delitem__(self, key)
|
||||
|
||||
def clear(self):
|
||||
self.callback(self.field)
|
||||
return dict.clear(self)
|
||||
|
||||
def pop(self, *args):
|
||||
self.callback(self.field)
|
||||
return dict.pop(self, *args)
|
||||
|
||||
def popitem(self):
|
||||
self.callback(self.field)
|
||||
return dict.popitem(self)
|
||||
|
||||
def setdefault(self, *args):
|
||||
self.callback(self.field)
|
||||
if len(args) == 2:
|
||||
return dict.setdefault(self, args[0], watch(
|
||||
args[1], callback=self.callback, field=self.field))
|
||||
else:
|
||||
return dict.setdefault(self, *args)
|
||||
|
||||
# XXX worth implementing?
|
||||
update = None
|
||||
|
||||
class WatchedList(list, object):
|
||||
def __init__(self, l, callback, field):
|
||||
self.callback = callback
|
||||
self.field = field
|
||||
for item in l:
|
||||
list.append(self, watch(item, callback=callback, field=self.field))
|
||||
|
||||
def __setitem__(self, index, value):
|
||||
self.callback(self.field)
|
||||
return list.__setitem__(self, index, watch(
|
||||
value, callback=self.callback, field=self.field))
|
||||
|
||||
def __delitem__(self, index):
|
||||
self.callback(self.field)
|
||||
return list.__delitem__(self, index)
|
||||
|
||||
def append(self, value):
|
||||
self.callback(self.field)
|
||||
return list.append(self, watch(
|
||||
value, callback=self.callback, field=self.field))
|
||||
|
||||
def extend(self, value):
|
||||
self.callback(self.field)
|
||||
return list.extend(self, watch(
|
||||
list(value), callback=self.callback, field=self.field))
|
||||
|
||||
def insert(self, index, value):
|
||||
self.callback(self.field)
|
||||
return list.insert(self, index, watch(
|
||||
value, callback=self.callback, field=self.field))
|
||||
|
||||
def remove(self, value):
|
||||
self.callback(self.field)
|
||||
return list.remove(self, value)
|
||||
|
||||
def pop(self, index=-1):
|
||||
self.callback(self.field)
|
||||
return list.pop(self, index)
|
||||
|
||||
# python 2.7 doesn't have this anyway
|
||||
clear = None
|
||||
|
||||
def sort(self, key=None, reverse=False):
|
||||
self.callback(self.field)
|
||||
return list.sort(self, key, reverse)
|
||||
|
||||
def reverse(self):
|
||||
self.callback(self.field)
|
||||
return list.reverse(self)
|
||||
|
||||
def watch(obj, callback, field):
|
||||
if isinstance(obj, dict):
|
||||
return WatchedDict(obj, callback, field)
|
||||
elif isinstance(obj, list):
|
||||
return WatchedList(obj, callback, field)
|
||||
else:
|
||||
return obj
|
||||
|
||||
class classproperty(object):
|
||||
def __init__(self, fget):
|
||||
self.fget = fget
|
||||
def __get__(self, owner_self, owner_cls):
|
||||
return self.fget(owner_cls)
|
||||
|
||||
class Document(dict, object):
|
||||
class Document(dict):
|
||||
'''
|
||||
Base class for ORM. You should subclass this class for each of your
|
||||
rethinkdb tables. You can add custom functionality in your subclass if
|
||||
appropriate.
|
||||
Base class for ORM.
|
||||
|
||||
This class keeps track of changes made to the object and any nested fields.
|
||||
After you have made some changes, call update() to persist them to the
|
||||
database.
|
||||
You should subclass this class for each of your rethinkdb tables. You can
|
||||
add custom functionality in your subclass if appropriate.
|
||||
|
||||
Changes in nested fields result in updates to their first-level ancestor
|
||||
field. For example, if your document starts as {'a': {'b': 'c'}}, then
|
||||
you run doc['a']['x'] = 'y', then the update will replace the whole 'a'
|
||||
field. (Nested field updates get too complicated any other way.)
|
||||
Call save() to persist changes to the model.
|
||||
|
||||
This class subclasses dict. Thus attributes can be accessed with
|
||||
`doc['foo']` or `doc.get('foo')`, depending on what you want to happen if
|
||||
@ -152,7 +48,6 @@ class Document(dict, object):
|
||||
class Something(doublethink.Document):
|
||||
table = 'my_table_name'
|
||||
'''
|
||||
|
||||
@classproperty
|
||||
def table(cls):
|
||||
return cls.__name__.lower()
|
||||
@ -215,55 +110,19 @@ class Document(dict, object):
|
||||
'''
|
||||
dict.__setattr__(self, 'rr', rr)
|
||||
self._pk = None
|
||||
self._clear_updates()
|
||||
for k in d or {}:
|
||||
dict.__setitem__(
|
||||
self, k, watch(d[k], callback=self._updated, field=k))
|
||||
self.update(d or {})
|
||||
self.populate_defaults()
|
||||
|
||||
def _clear_updates(self):
|
||||
self._updates = {}
|
||||
self._deletes = set()
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
# keys starting with underscore are not part of the document
|
||||
if key[:1] == '_':
|
||||
dict.__setattr__(self, key, value)
|
||||
else:
|
||||
dict.__setitem__(
|
||||
self, key, watch(value, callback=self._updated, field=key))
|
||||
self._updated(key)
|
||||
dict.__setitem__(self, key, value)
|
||||
|
||||
__setattr__ = __setitem__
|
||||
__getattr__ = dict.get
|
||||
|
||||
def __delitem__(self, key):
|
||||
dict.__delitem__(self, key)
|
||||
self._deletes.add(key)
|
||||
if key in self._updates:
|
||||
del self._updates[key]
|
||||
|
||||
def setdefault(self, *args):
|
||||
need_update = False
|
||||
if not args[0] in self:
|
||||
need_update = True
|
||||
result = dict.setdefault(self, *args)
|
||||
if need_update:
|
||||
self._updated(args[0])
|
||||
return result
|
||||
|
||||
# dict methods we don't want to support
|
||||
clear = None
|
||||
pop = None
|
||||
popitem = None
|
||||
update = None
|
||||
|
||||
def _updated(self, field):
|
||||
# callback for all updates
|
||||
self._updates[field] = self[field]
|
||||
if field in self._deletes:
|
||||
self._deletes.remove(field)
|
||||
|
||||
@property
|
||||
def pk_field(self):
|
||||
'''
|
||||
@ -293,73 +152,27 @@ class Document(dict, object):
|
||||
|
||||
def populate_defaults(self):
|
||||
'''
|
||||
This method is called by `save()` before persisting the document to
|
||||
the database. Subclasses should override it to populate default values
|
||||
if appropriate.
|
||||
This method is called by `__init__()`. Subclasses should override it to
|
||||
populate default values if appropriate.
|
||||
'''
|
||||
pass
|
||||
|
||||
def save(self):
|
||||
'''
|
||||
Persist changes to rethinkdb. Updates only the fields that have
|
||||
changed. Performs insert rather than update if the document has no
|
||||
primary key or if the primary key is absent from the database.
|
||||
|
||||
If there have been any changes to nested fields, updates the first
|
||||
level attribute. For example, if foo['bar']['baz']['quux'] has changed,
|
||||
all of foo['bar'] is replaced, but foo['something_else'] is not
|
||||
touched.
|
||||
'''
|
||||
should_insert = False
|
||||
try:
|
||||
self[self.pk_field] # raises KeyError if missing
|
||||
if self._updates:
|
||||
# r.literal() to replace, not merge with, nested fields
|
||||
updates = {field: r.literal(self._updates[field])
|
||||
for field in self._updates}
|
||||
query = self.rr.table(self.table).get(
|
||||
self.pk_value).update(updates)
|
||||
result = query.run()
|
||||
if result['skipped']: # primary key not found
|
||||
should_insert = True
|
||||
elif result['errors'] or result['deleted']:
|
||||
raise Exception(
|
||||
'unexpected result %s from rethinkdb query %s' % (
|
||||
result, query))
|
||||
if not should_insert and self._deletes:
|
||||
query = self.rr.table(self.table).get(self.pk_value).replace(
|
||||
r.row.without(self._deletes))
|
||||
result = query.run()
|
||||
if result['errors']: # primary key not found
|
||||
should_insert = True
|
||||
elif result['replaced'] != 1:
|
||||
raise Exception(
|
||||
'unexpected result %s from rethinkdb query %s' % (
|
||||
result, query))
|
||||
except KeyError:
|
||||
should_insert = True
|
||||
|
||||
if should_insert:
|
||||
query = self.rr.table(self.table).insert(self)
|
||||
result = query.run()
|
||||
if result['inserted'] != 1:
|
||||
raise Exception(
|
||||
'unexpected result %s from rethinkdb query %s' % (
|
||||
result, query))
|
||||
if 'generated_keys' in result:
|
||||
dict.__setitem__(
|
||||
self, self.pk_field, result['generated_keys'][0])
|
||||
|
||||
self._clear_updates()
|
||||
'''Persist changes to rethinkdb.'''
|
||||
query = self.rr.table(self.table).insert(self, conflict='replace')
|
||||
result = query.run()
|
||||
if sorted([result['inserted'], result['replaced'], result['unchanged']]) != [0,0,1]:
|
||||
raise Exception(
|
||||
'unexpected result %s from rethinkdb query %s' % (
|
||||
result, query))
|
||||
if 'generated_keys' in result:
|
||||
self[self.pk_field] = result['generated_keys'][0]
|
||||
|
||||
def refresh(self):
|
||||
'''
|
||||
Refresh the document from the database.
|
||||
'''
|
||||
d = self.rr.table(self.table).get(self.pk_value).run()
|
||||
if d is None:
|
||||
raise KeyError
|
||||
for k in d:
|
||||
dict.__setitem__(
|
||||
self, k, watch(d[k], callback=self._updated, field=k))
|
||||
self.clear()
|
||||
self.update(d)
|
||||
|
||||
|
||||
@ -53,7 +53,7 @@ class RethinkerWrapper(object):
|
||||
'^Cannot perform.*replica.*', e.args[0]):
|
||||
if error_count < 20:
|
||||
error_count += 1
|
||||
self.logger.warn(
|
||||
self.logger.warning(
|
||||
'will keep trying after potentially '
|
||||
'recoverable error (%s/20): %s',
|
||||
error_count, e)
|
||||
@ -90,7 +90,7 @@ class RethinkerWrapper(object):
|
||||
except r.ReqlOpFailedError as e:
|
||||
if e.args and re.match(
|
||||
'^Cannot perform.*replica.*', e.args[0]):
|
||||
self.logger.warn(
|
||||
self.logger.warning(
|
||||
'will keep trying after potentially recoverable '
|
||||
'error: %s', e)
|
||||
time.sleep(0.5)
|
||||
@ -129,7 +129,7 @@ class Rethinker(object):
|
||||
return r.connect(host=server)
|
||||
except Exception as e:
|
||||
self.last_error[server] = time.time()
|
||||
self.logger.warn(
|
||||
self.logger.warning(
|
||||
'will keep trying after failure connecting to '
|
||||
'rethinkdb server at %s: %s (sleeping for %s sec)',
|
||||
server, e, retry_wait)
|
||||
|
||||
@ -168,7 +168,7 @@ class ServiceRegistry(object):
|
||||
if result != {
|
||||
'deleted':1, 'errors':0,'inserted':0,
|
||||
'replaced':0,'skipped':0,'unchanged':0}:
|
||||
self.logger.warn(
|
||||
self.logger.warning(
|
||||
'unexpected result attempting to delete id=%s from '
|
||||
'rethinkdb services table: %s', id, result)
|
||||
|
||||
|
||||
@ -66,117 +66,31 @@ def test_orm(rr):
|
||||
'i': ['j', {'k': 'l'}]})
|
||||
d.save()
|
||||
|
||||
assert d._updates == {}
|
||||
d.m = 'n'
|
||||
assert d._updates == {'m': 'n'}
|
||||
d['c']['o'] = 'p'
|
||||
assert d._updates == {'m': 'n', 'c': {'d': 'e', 'o': 'p'}}
|
||||
d.f[0] = 'q'
|
||||
assert d._updates == {'m': 'n', 'c': {'d': 'e', 'o': 'p'}, 'f': ['q', 'h']}
|
||||
d['i'][1]['k'] = 's'
|
||||
assert d._updates == {
|
||||
'm': 'n',
|
||||
'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['q', 'h'],
|
||||
'i': ['j', {'k': 's'}]}
|
||||
|
||||
del d['i']
|
||||
assert d._deletes == {'i'}
|
||||
assert d._updates == {'m': 'n', 'c': {'d': 'e', 'o': 'p'}, 'f': ['q', 'h']}
|
||||
|
||||
d.i = 't'
|
||||
assert d._deletes == set()
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'}, 'f': ['q', 'h'], 'i': 't'}
|
||||
|
||||
# list manipulations
|
||||
d.f.append(['sublist'])
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['q', 'h', ['sublist']], 'i': 't'}
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
d.f[2].clear()
|
||||
|
||||
result = d.f.pop()
|
||||
assert result == ['sublist']
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['q', 'h'], 'i': 't'}
|
||||
|
||||
del d.f[0]
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['h'], 'i': 't'}
|
||||
|
||||
d.f.insert(0, 'u')
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['u', 'h'], 'i': 't'}
|
||||
|
||||
d.f.extend(('v', {'w': 'x'}))
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['u', 'h', 'v', {'w': 'x'}], 'i': 't'}
|
||||
|
||||
# check that stuff added by extend() is watched properly
|
||||
d.f[3]['y'] = 'z'
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['u', 'h', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
d.f.remove('h')
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'d': 'e', 'o': 'p'},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
# more nested field dict operations
|
||||
del d['c']['d']
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'o': 'p'},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
d['c'].clear()
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
assert d['c'].setdefault('aa') is None
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'aa': None},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
d['c'].setdefault('aa', 'bb') is None
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'aa': None},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
d['c'].setdefault('cc', 'dd') == 'dd'
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'aa': None, 'cc': 'dd'},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
d['c'].setdefault('cc') == 'dd'
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'aa': None, 'cc': 'dd'},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
d['c'].setdefault('cc', 'ee') == 'dd'
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'aa': None, 'cc': 'dd'},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
assert d['c'].pop('cc') == 'dd'
|
||||
assert d._updates == {
|
||||
'm': 'n', 'c': {'aa': None},
|
||||
'f': ['u', 'v', {'w': 'x', 'y': 'z'}], 'i': 't'}
|
||||
|
||||
assert d['f'][2].popitem()
|
||||
assert d._updates['f'][2] in ({'w':'x'}, {'y':'z'})
|
||||
|
||||
d.save()
|
||||
assert d._updates == {}
|
||||
assert d._deletes == set()
|
||||
|
||||
d_copy = SomeDoc.load(rr, d.id)
|
||||
assert d == d_copy
|
||||
@ -186,37 +100,12 @@ def test_orm(rr):
|
||||
d_copy.refresh()
|
||||
assert d == d_copy
|
||||
|
||||
# top level dict operations
|
||||
with pytest.raises(TypeError):
|
||||
d.clear()
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
d.pop('m')
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
d.popitem()
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
d.update({'x':'y'})
|
||||
|
||||
assert d.setdefault('ee') is None
|
||||
assert d._updates == {'ee': None}
|
||||
|
||||
d.setdefault('ee', 'ff') is None
|
||||
assert d._updates == {'ee': None}
|
||||
|
||||
d.setdefault('gg', 'hh') == 'hh'
|
||||
assert d._updates == {'ee': None, 'gg': 'hh'}
|
||||
|
||||
d.setdefault('gg') == 'hh'
|
||||
assert d._updates == {'ee': None, 'gg': 'hh'}
|
||||
|
||||
d.setdefault('gg', 'ii') == 'hh'
|
||||
assert d._updates == {'ee': None, 'gg': 'hh'}
|
||||
|
||||
d.save()
|
||||
assert d._updates == {}
|
||||
assert d._deletes == set()
|
||||
|
||||
d_copy = SomeDoc.load(rr, d.id)
|
||||
assert d == d_copy
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user