api: add limit/offset/sort to api/users/<...>/forms (#35403)

tests/test_api.py

@@ -1385,6 +1385,65 @@ def test_user_forms(pub, local_user):
     draft_formdata = [x for x in resp.json['data'] if x['status'] == 'Draft'][0]
     assert draft_formdata.get('url')[-1] != '/'
 
+    formdata = formdef.data_class()()
+    formdata.data = {'0': 'foo@localhost', '1': 'xyy'}
+    formdata.user_id = local_user.id
+    formdata.just_created()
+    formdata.jump_status('new')
+    formdata.store()
+
+    resp = get_app(pub).get(sign_uri('/api/user/forms', user=local_user))
+    assert len(resp.json['data']) == 2
+    resp2 = get_app(pub).get(sign_uri('/api/user/forms?sort=desc', user=local_user))
+    assert len(resp2.json['data']) == 2
+    assert resp2.json['data'][0] == resp.json['data'][1]
+    assert resp2.json['data'][1] == resp.json['data'][0]
+
+def test_user_forms_limit_offset(pub, local_user):
+    if not pub.is_using_postgresql():
+        pytest.skip('this requires SQL')
+        return
+
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.name = 'test limit offset'
+    formdef.fields = [
+        fields.StringField(id='0', label='foobar', varname='foobar'),
+        fields.StringField(id='1', label='foobar2'),]
+    formdef.keywords = 'hello, world'
+    formdef.disabled = False
+    formdef.enable_tracking_codes = False
+    formdef.store()
+    formdef.data_class().wipe()
+
+    for i in range(50):
+        formdata = formdef.data_class()()
+        formdata.data = {'0': 'foo@localhost', '1': str(i)}
+        formdata.user_id = local_user.id
+        formdata.just_created()
+        formdata.receipt_time = (datetime.datetime.now() + datetime.timedelta(days=i)).timetuple()
+        formdata.jump_status('new')
+        formdata.store()
+
+    resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id))
+    assert resp.json['err'] == 0
+    assert len(resp.json['data']) == 50
+
+    resp = get_app(pub).get(sign_uri('/api/users/%s/forms?limit=10' % local_user.id))
+    assert resp.json['err'] == 0
+    assert len(resp.json['data']) == 10
+    assert [x['form_number_raw'] for x in resp.json['data']] == [str(x) for x in range(1, 11)]
+
+    resp = get_app(pub).get(sign_uri('/api/users/%s/forms?limit=10&offset=45' % local_user.id))
+    assert resp.json['err'] == 0
+    assert len(resp.json['data']) == 5
+    assert [x['form_number_raw'] for x in resp.json['data']] == [str(x) for x in range(46, 51)]
+
+    resp = get_app(pub).get(sign_uri('/api/users/%s/forms?limit=10&sort=desc' % local_user.id))
+    assert resp.json['err'] == 0
+    assert len(resp.json['data']) == 10
+    assert [x['form_number_raw'] for x in resp.json['data']] == [str(x) for x in range(50, 40, -1)]
+
 def test_user_forms_from_agent(pub, local_user):
     Role.wipe()
     role = Role(name='Foo bar')

wcs/api.py

@@ -618,10 +618,18 @@ class ApiUserDirectory(Directory):
         if FormDef.count() == 0:
             # early return, this avoids running a query against a missing SQL view.
             return []
+
         if get_publisher().is_using_postgresql() and not get_request().form.get('full') == 'on':
             from wcs import sql
             from qommon.storage import Equal
-            user_forms = sql.AnyFormData.select([Equal('user_id', str(user.id))], order_by='receipt_time')
+            order_by = 'receipt_time'
+            if get_request().form.get('sort') == 'desc':
+                order_by = '-receipt_time'
+            user_forms = sql.AnyFormData.select(
+                    [Equal('user_id', str(user.id))],
+                    limit=get_request().form.get('limit'),
+                    offset=get_request().form.get('offset'),
+                    order_by=order_by)
         else:
             formdefs = FormDef.select()
             user_forms = []
@@ -629,6 +637,8 @@ class ApiUserDirectory(Directory):
                 user_forms.extend(formdef.data_class().get_with_indexed_value(
                             'user_id', user.id))
             user_forms.sort(lambda x, y: cmp(x.receipt_time, y.receipt_time))
+            if get_request().form.get('sort') == 'desc':
+                user_forms.reverse()
         return user_forms
 
     def drafts(self):