entrouvert/wcs
entrouvert
/
wcs
14
1
Fork 0
Code Issues Pull Requests 35 Releases Activity

misc: improve |getlist filter for querysets (#49864)

This commit is contained in:
Lauréline Guérin 2021-01-21 15:53:50 +01:00
parent 45ef92e75d
commit 2660c62368
No known key found for this signature in database
GPG Key ID: 1FAB9B9B4F93D473
3 changed files with 109 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
tests/test_formdata.py
View File

@ -1222,16 +1222,30 @@ def test_lazy_formdata_queryset_filter(pub, variable_test_data):
# test |getlist
tmpl = Template('{% for v in form_objects|order_by:"id"|getlist:"foo_foo" %}{{ v }},{% endfor %}')
assert tmpl.render(context) == 'bar,bar,bar,bar,bar,bar,bar,foo,foo,foo,foo,'
tmpl = Template('{% for v in form_objects|order_by:"id"|getlist:"datefield" %}{{ v|date }},{% endfor %}')
assert tmpl.render(context) == '2018-07-31,,,,,,,2018-07-31,2018-07-31,2018-07-31,2018-07-31,'
tmpl = Template('{% if "foo" in form_objects|getlist:"foo_foo" %}OK{% else %}KO{% endif%}')
assert tmpl.render(context) == 'OK'
tmpl = Template('{% if "fooooooooooooooo" in form_objects|getlist:"foo_foo" %}OK{% else %}KO{% endif%}')
assert tmpl.render(context) == 'KO'
tmpl = Template('{{ form_objects|order_by:"id"|getlist:"foo_foo"|count }}')
assert tmpl.render(context) == '11'
assert LazyFormData(formdata).objects.order_by('id').getlist('foo_foo') == [
'bar', 'bar', 'bar', 'bar', 'bar', 'bar', 'bar', 'foo', 'foo', 'foo', 'foo']
assert LazyFormData(formdata).objects.order_by('id').getlist('form_var_foo_foo') == [
'bar', 'bar', 'bar', 'bar', 'bar', 'bar', 'bar', 'foo', 'foo', 'foo', 'foo']
assert set(LazyFormData(formdata).objects.getlist('unknown')) == set([None])
# test with cache populated
for value in [datetime.date(2018, 7, 31).timetuple(), datetime.date(2018, 7, 31), datetime.datetime(2018, 7, 31), '2018-07-31']:
assert value in LazyFormData(formdata).objects.getlist('datefield')
tmpl = Template('''{% with form_objects|getlist:"datefield" as objects %}
{% for v in objects %}{% endfor %}{% if value in objects %}OK{% else %}KO{% endif%}
{% endwith %}''')
context['value'] = value
assert tmpl.render(context) == '\n OK\n '
assert 'not a date' not in LazyFormData(formdata).objects.getlist('datefield')
def test_lazy_formdata_queryset_get_from_first(pub, variable_test_data):
context = pub.substitutions.get_context_variables(mode='lazy')

1
tests/test_formdef.py
View File

@ -189,6 +189,7 @@ def test_substitution_variables_object(pub):
d = formdef.data_class()()
d.status = 'wf-1'
d.store()
substs = formdef.get_substitution_variables().get('form_objects')
assert substs.count == 1
assert substs.count_status_1 == 1

135
wcs/variables.py
View File

@ -35,7 +35,7 @@ from .formdef import FormDef
class LazyFormDefObjectsManager(object):
def __init__(self, formdef, formdata=None, geoloc_center_formdata=None, criterias=None, order_by=None):
def __init__(self, formdef, formdata=None, geoloc_center_formdata=None, criterias=None, order_by=None, limit=None):
self._formdef = formdef
self._formdata = formdata
self._geoloc_center_formdata = geoloc_center_formdata
@ -50,11 +50,14 @@ class LazyFormDefObjectsManager(object):
criterias[1].exclude_anonymised = True
self._criterias = criterias
self._order_by = order_by
self._limit = limit
self._cached_resultset = None
@property # @property for backward compatibility
def count(self):
return self._formdef.data_class().count(clause=self._criterias)
if not hasattr(self, '_count_cache'):
self._count_cache = self._formdef.data_class().count(clause=self._criterias)
return self._count_cache
def _clone(self, criterias, order_by=None):
return LazyFormDefObjectsManager(
@ -62,11 +65,17 @@ class LazyFormDefObjectsManager(object):
formdata=self._formdata,
geoloc_center_formdata=self._geoloc_center_formdata,
criterias=criterias,
order_by=order_by or self._order_by)
order_by=order_by or self._order_by,
limit=self._limit)
def order_by(self, attribute):
return self._clone(self._criterias, order_by=attribute)
def limit(self, limit):
qs = self._clone(self._criterias)
qs._limit = limit
return qs
def all(self):
# (expose 'all' only to mimick django, it's not actually useful as this
# object serves as both manager and queryset)
@ -161,37 +170,42 @@ class LazyFormDefObjectsManager(object):
qs.pending_attr = attribute
return qs
def format_value(self, field, value):
if field.key != 'date':
return value
return parse_date(value).timetuple()
def get_field(self, key):
for field in self._formdef.get_all_fields():
if getattr(field, 'varname', None) == key:
return field
def apply_filter_value(self, value):
assert self.pending_attr
def format_value(field, value):
if field.key != 'date':
return value
return parse_date(value).timetuple()
field = self.get_field(self.pending_attr)
if field is None:
get_publisher().record_error(_('Invalid filter "%s"') % self.pending_attr, formdata=self._formdata)
return self.none()
for field in self._formdef.get_all_fields():
if getattr(field, 'varname', None) == self.pending_attr:
try:
value = format_value(field, value)
except (ValueError, AttributeError):
get_publisher().record_error(_('Invalid value "%s" for filter "%s"') % (value, self.pending_attr), formdata=self._formdata)
return self.none()
try:
value = self.format_value(field, value)
except (ValueError, AttributeError):
get_publisher().record_error(_('Invalid value "%s" for filter "%s"') % (value, self.pending_attr), formdata=self._formdata)
return self.none()
from wcs import sql
criteria = Equal(sql.get_field_id(field), value)
return self._clone(self._criterias + [criteria])
get_publisher().record_error(_('Invalid filter "%s"') % self.pending_attr, formdata=self._formdata)
return self.none()
from wcs import sql
criteria = Equal(sql.get_field_id(field), value)
return self._clone(self._criterias + [criteria])
def getlist(self, key):
values = []
for lazy_formdata in self:
value = lazy_formdata.get(key)
if value is not None:
value = value.get_value()
values.append(value)
return values
return LazyList(self, key)
def _populate_cache(self):
if self._cached_resultset is not None:
return
result = self._formdef.data_class().select(clause=self._criterias, order_by=self._order_by, limit=self._limit)
self._cached_resultset = [LazyFormData(x) for x in result]
def __getattr__(self, attribute):
if attribute.startswith('count_status_'):
@ -208,7 +222,7 @@ class LazyFormDefObjectsManager(object):
raise AttributeError('No such attribute %r' % attribute)
def __len__(self):
if self._cached_resultset:
if self._cached_resultset is not None:
return len(self._cached_resultset)
return self.count
@ -223,28 +237,67 @@ class LazyFormDefObjectsManager(object):
# We need to abort earlier as we don't want to load all formdata
# in that situation.
raise TypeError
if self._cached_resultset is None:
self._cached_resultset = [LazyFormData(x) for x in
self._formdef.data_class().select(clause=self._criterias, order_by=self._order_by)]
self._populate_cache()
return self._cached_resultset[key]
def __iter__(self):
if self._cached_resultset:
for lazy_formdata in self._cached_resultset:
yield lazy_formdata
data_class = self._formdef.data_class()
temp_cached_resultset = []
for formdata in self._formdef.data_class().select_iterator(
clause=self._criterias, order_by=self._order_by):
lazy_formdata = LazyFormData(formdata)
temp_cached_resultset.append(lazy_formdata)
self._populate_cache()
for lazy_formdata in self._cached_resultset:
yield lazy_formdata
self._cached_resultset = temp_cached_resultset
def __nonzero__(self):
return any(self)
class LazyList(object):
def __init__(self, lazy_manager, key):
self._lazy_manager = lazy_manager
self._key = key
self._cached_resultset = None
def _populate_cache(self):
if self._cached_resultset is not None:
return
self._cached_resultset = []
for lazy_formdata in self._lazy_manager:
value = lazy_formdata.get(self._key)
if value is not None:
if hasattr(value, 'timetuple'):
value = value.timetuple()
else:
value = value.get_value()
self._cached_resultset.append(value)
def __len__(self):
if self._cached_resultset is not None:
return len(self._cached_resultset)
return len(self._lazy_manager)
def __iter__(self):
self._populate_cache()
for value in self._cached_resultset:
yield value
def __nonzero__(self):
return any(self)
def __contains__(self, value):
if self._cached_resultset is not None:
field = self._lazy_manager.get_field(self._key)
if field is not None:
try:
value = self._lazy_manager.format_value(field, value)
except (ValueError, AttributeError):
pass
return value in list(self)
queryset = list(self._lazy_manager.filter_by(self._key).apply_filter_value(value).limit(1))
return len(queryset) == 1
def __eq__(self, other):
return list(self) == list(other)
class LazyFormDef(object):
def __init__(self, formdef):
self._formdef = formdef