0001-api-add-filter-operators-on-formdata-listing-60785.patch

Lauréline Guérin, 31 janvier 2022 14:00

Voir les différences: en ligne côte à côte

Subject: [PATCH] api: add filter operators on formdata listing (#60785)

 tests/api/test_formdata.py   | 632 ++++++++++++++++++++++++++++++++++-
 wcs/backoffice/management.py | 108 +++---
 wcs/qommon/storage.py        |  22 +-
 wcs/sql.py                   |  70 +++-
 4 files changed, 777 insertions(+), 55 deletions(-)

         assert resp.json['err_desc'] == 'Invalid filters "baz", "foobar"'
     def test_api_list_formdata_string_filter(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = [
             fields.StringField(id='0', label='String', type='string', varname='string'),
             fields.StringField(id='1', label='String2', type='string', varname='string2'),
+        ]
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(3):
             formdata = data_class()
             formdata.data = {
                 '0': 'FOO %s' % i,
                 '1': '%s' % (9 + i),
+            }
             formdata.user_id = local_user.id
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-string=FOO 2', user=local_user))
         assert len(resp.json) == 1
         params = [
             ('eq', 'FOO 2', 1),
             ('ne', 'FOO 2', 2),
             ('lt', 'FOO 2', 2),
             ('lte', 'FOO 2', 3),
             ('gt', 'FOO 2', 0),
             ('gt', '42', 0),
             ('gte', 'FOO 2', 1),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-string=%s&filter-string-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         params = [
             ('eq', '10', 1),
             ('eq', '010', 1),
             ('ne', '10', 2),
             ('lt', '10', 1),
             ('lte', '10', 2),
             ('gt', '10', 1),
             ('gt', '9', 2),
             ('gte', '10', 2),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-string2=%s&filter-string2-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
     def test_api_list_formdata_item_filter(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         NamedDataSource.wipe()
         data_source = NamedDataSource(name='foobar')
         data_source.data_source = {
             'type': 'formula',
             'value': repr([{'id': '9', 'text': 'foo'}, {'id': '10', 'text': 'bar'}, {'id': '11', 'text': 'baz'}]),
+        }
         data_source.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = [
             fields.ItemField(id='0', label='Item', type='item', data_source={'type': 'foobar'}, varname='item'),
             fields.ItemField(id='1', label='Other Item', type='item', items=['foo', 'bar'], varname='item2'),
+        ]
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(3):
             formdata = data_class()
             formdata.data = {
                 '0': str(9 + i),
                 '1': 'foo' if i % 2 else 'bar',
+            }
             formdata.user_id = local_user.id
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-item=9', user=local_user))
         assert len(resp.json) == 1
         params = [
             ('eq', '10', 1),
             ('eq', '010', 1),
             ('ne', '10', 2),
             ('lt', '10', 1),
             ('lte', '10', 2),
             ('gt', '10', 1),
             ('gt', '9', 2),
             ('gte', '10', 2),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-item=%s&filter-item-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-item2=foo', user=local_user))
         assert len(resp.json) == 1
         params = [
             ('eq', 'foo', 1),
             ('ne', 'foo', 2),
             ('lt', 'foo', 2),
             ('lte', 'foo', 3),
             ('gt', 'foo', 0),
             ('gt', '42', 0),
             ('gte', 'foo', 1),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-item2=%s&filter-item2-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
     def test_api_list_formdata_items_filter(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         NamedDataSource.wipe()
         data_source = NamedDataSource(name='foobar')
         data_source.data_source = {
             'type': 'formula',
             'value': repr([{'id': '9', 'text': 'foo'}, {'id': '10', 'text': 'bar'}, {'id': '11', 'text': 'baz'}]),
+        }
         data_source.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = [
             fields.ItemsField(
                 id='0', label='Items', type='items', data_source={'type': 'foobar'}, varname='items'
             ),
             fields.ItemsField(
                 id='1', label='Other Item', type='items', items=['foo', 'bar', 'baz'], varname='items2'
             ),
+        ]
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(4):
             formdata = data_class()
             formdata.data = {}
             if i < 3:
                 formdata.data = {
                     '0': ['9' if i % 2 else '11', '10'],
                     '1': ['foo' if i % 2 else 'bar', 'baz'],
+                }
             formdata.user_id = local_user.id
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-items=11', user=local_user))
         assert len(resp.json) == 2
         params = [
             ('eq', '11', 2),
             ('eq', '011', 2),
             ('eq', '10', 3),
             ('ne', '9', 3),
             ('ne', '10', 1),
             ('lt', '10', 1),
             ('lte', '10', 3),
             ('gt', '10', 2),
             ('gt', '9', 3),
             ('gte', '11', 2),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-items=%s&filter-items-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         params = [
             ('eq', 'foo', 1),
             ('ne', 'foo', 3),
             ('lt', 'foo', 3),
             ('lte', 'foo', 3),
             ('gt', 'foo', 0),
             ('gt', '42', 0),
             ('gte', 'foo', 1),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-items2=%s&filter-items2-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
     def test_api_list_formdata_bool_filter(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = [
             fields.BoolField(id='0', label='Bool', type='bool', varname='bool'),
+        ]
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(3):
             formdata = data_class()
             formdata.data = {
                 '0': bool(i % 2),
+            }
             formdata.user_id = local_user.id
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-bool=false', user=local_user))
         assert len(resp.json) == 2
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-bool=true', user=local_user))
         assert len(resp.json) == 1
         params = [
             ('eq', 'true', 1),
             ('ne', 'true', 2),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-bool=%s&filter-bool-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         for operator in ['lt', 'lte', 'gt', 'gte']:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-bool=true&filter-bool-operator=%s' % operator, user=local_user
                 ),
                 status=400,
+            )
             assert resp.json['err_desc'] == 'Invalid operator "%s" for "filter-bool"' % operator
     def test_api_list_formdata_date_filter(pub, local_user):
         if not pub.is_using_postgresql():
             pytest.skip('this requires SQL')
-...
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = [
             fields.DateField(id='0', label='foobar', varname='foobar', type='date'),
             fields.DateField(id='0', label='Date', varname='date', type='date'),
+        ]
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(30):
         for i in range(3):
             formdata = data_class()
             formdata.data = {'0': time.strptime('2021-06-%02d' % (i + 1), '%Y-%m-%d')}
             formdata.data = {'0': time.strptime('2021-06-%02d' % (i + 10), '%Y-%m-%d')}
             formdata.user_id = local_user.id
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-foobar=2021-06-12', user=local_user))
         assert len(resp.json) == 1
         for value in ['2021-06-11', '11/06/2021']:
             resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-date=%s' % value, user=local_user))
             assert len(resp.json) == 1
             params = [
                 ('eq', 1),
                 ('ne', 2),
                 ('lt', 1),
                 ('lte', 2),
                 ('gt', 1),
                 ('gte', 2),
+            ]
             for operator, result in params:
                 resp = get_app(pub).get(
                     sign_uri(
                         '/api/forms/test/list?filter-date=%s&filter-date-operator=%s' % (value, operator),
                         user=local_user,
+                    )
+                )
                 assert len(resp.json) == result
         # alternate date format
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-foobar=12/06/2021', user=local_user))
     def test_api_list_formdata_email_filter(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = [
             fields.EmailField(id='0', label='Email', type='email', varname='email'),
+        ]
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(3):
             formdata = data_class()
             formdata.data = {'0': 'a@localhost' if i % 2 else 'b@localhost'}
             formdata.user_id = local_user.id
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-email=a@localhost', user=local_user))
         assert len(resp.json) == 1
         params = [
             ('eq', 'a@localhost', 1),
             ('ne', 'a@localhost', 2),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-email=%s&filter-email-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         for operator in ['lt', 'lte', 'gt', 'gte']:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-email=a@localhost&filter-email-operator=%s' % operator,
                     user=local_user,
                 ),
                 status=400,
+            )
             assert resp.json['err_desc'] == 'Invalid operator "%s" for "filter-email"' % operator
     def test_api_list_formdata_internal_id_filter(pub, local_user):
-...
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(2):
         for i in range(11):
             formdata = data_class()
             formdata.data = {}
             formdata.user_id = local_user.id
-...
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-internal-id=42', user=local_user))
         assert len(resp.json) == 0
         params = [
             ('eq', '1', 1),
             ('eq', '01', 1),
             ('ne', '1', 10),
             ('lt', '1', 0),
             ('lte', '1', 1),
             ('gt', '1', 10),
             ('gt', '10', 1),
             ('gte', '1', 11),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-internal-id=%s&filter-internal-id-operator=%s'
                     % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         resp = get_app(pub).get(
             sign_uri('/api/forms/test/list?filter-internal-id=blabla', user=local_user), status=400
+        )
         assert resp.json['err_desc'] == 'Invalid value "blabla" for "filter-internal-id-value"'
     def test_api_list_formdata_user_filter(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         user1 = pub.user_class(name='userA')
         user1.name_identifiers = ['ABCDEF']
         user1.store()
         user2 = pub.user_class(name='userB')
         user1.name_identifiers = ['GHIJKL']
         user2.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = []
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(3):
             formdata = data_class()
             formdata.data = {}
             formdata.user_id = str(user1.id if i % 2 else user2.id)
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-user-uuid=ABCDEF', user=local_user))
         assert len(resp.json) == 1
         params = [
             ('eq', 'ABCDEF', 1),
             ('ne', 'ABCDEF', 2),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-user-uuid=%s&filter-user-uuid-operator=%s' % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         for operator in ['lt', 'lte', 'gt', 'gte']:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-user-uuid=ABCDEF&filter-user-uuid-operator=%s' % operator,
                     user=local_user,
                 ),
                 status=400,
+            )
             assert resp.json['err_desc'] == 'Invalid operator "%s" for "filter-user-value"' % operator
     def test_api_list_formdata_submission_agent_filter(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         local_user.roles = [role.id]
         local_user.store()
         user1 = pub.user_class(name='userA')
         user1.name_identifiers = ['ABCDEF']
         user1.store()
         user2 = pub.user_class(name='userB')
         user1.name_identifiers = ['GHIJKL']
         user2.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.workflow_roles = {'_receiver': role.id}
         formdef.fields = []
         formdef.store()
         data_class = formdef.data_class()
         data_class.wipe()
         for i in range(3):
             formdata = data_class()
             formdata.data = {}
             formdata.submission_agent_id = str(user1.id if i % 2 else user2.id)
             formdata.user_id = local_user.id
             formdata.just_created()
             formdata.jump_status('new')
             formdata.store()
         resp = get_app(pub).get(
             sign_uri('/api/forms/test/list?filter-submission-agent-uuid=ABCDEF', user=local_user)
+        )
         assert len(resp.json) == 1
         params = [
             ('eq', 'ABCDEF', 1),
             ('ne', 'ABCDEF', 2),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-submission-agent-uuid=%s&filter-submission-agent-uuid-operator=%s'
                     % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         for operator in ['lt', 'lte', 'gt', 'gte']:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-submission-agent-uuid=ABCDEF&filter-submission-agent-uuid-operator=%s'
                     % operator,
                     user=local_user,
                 ),
                 status=400,
+            )
             assert resp.json['err_desc'] == 'Invalid operator "%s" for "filter-submission-agent-value"' % operator
     def test_api_list_formdata_number_filter(pub, local_user):
         pub.role_class.wipe()
-...
         data_source = NamedDataSource(name='foobar')
         data_source.data_source = {
             'type': 'formula',
             'value': repr([{'id': '1', 'text': 'foo', 'more': 'XXX'}, {'id': '2', 'text': 'bar', 'more': 'YYY'}]),
             'value': repr([{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]),
+        }
         data_source.store()
-...
             if i == 0:
                 formdata.data['0']['data'].append(
+                    {
                         '1': 'plop%s' % i,
                         '1': 'plop%s' % (i + 1),
                         '2': '1',
                         '2_display': 'foo',
                         '2_structured': 'XXX',
-...
         assert len(resp.json) == 1
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-blockdata_string=plop10', user=local_user))
         assert len(resp.json) == 0
         params = [
             ('eq', 'plop5', 1),
             ('ne', 'plop5', 9),
             ('ne', 'plop1', 8),
             ('lt', 'plop5', 5),
             ('lte', 'plop5', 6),
             ('gt', 'plop5', 4),
             ('gt', '42', 0),
             ('gte', 'plop5', 5),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-blockdata_string=%s&filter-blockdata_string-operator=%s'
                     % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         # item
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-blockdata_item=1', user=local_user))
         assert len(resp.json) == 6
-...
         assert len(resp.json) == 5
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-blockdata_item=3', user=local_user))
         assert len(resp.json) == 0
         params = [
             ('eq', '1', 6),
             ('ne', '1', 4),
             ('lt', '2', 6),
             ('lte', '1', 6),
             ('gt', '1', 5),
             ('gte', '2', 5),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-blockdata_item=%s&filter-blockdata_item-operator=%s'
                     % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         # bool
         resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-blockdata_bool=true', user=local_user))
         assert len(resp.json) == 6
-...
             sign_uri('/api/forms/test/list?filter-blockdata_bool=foobar', user=local_user), status=400
+        )
         assert resp.json['err_desc'] == 'Invalid value "foobar" for "filter-blockdata_bool"'
         params = [
             ('eq', 'true', 6),
             ('ne', 'true', 4),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-blockdata_bool=%s&filter-blockdata_bool-operator=%s'
                     % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         for operator in ['lt', 'lte', 'gt', 'gte']:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-blockdata_bool=true&filter-blockdata_bool-operator=%s'
                     % operator,
                     user=local_user,
                 ),
                 status=400,
+            )
             assert resp.json['err_desc'] == 'Invalid operator "%s" for "filter-blockdata_bool"' % operator
         # date
         resp = get_app(pub).get(
             sign_uri('/api/forms/test/list?filter-blockdata_date=2021-06-01', user=local_user)
-...
             sign_uri('/api/forms/test/list?filter-blockdata_date=02/06/2021', user=local_user)
+        )
         assert len(resp.json) == 2
         params = [
             ('eq', '2021-06-02', 2),
             ('ne', '2021-06-02', 8),
             ('lt', '2021-06-02', 1),
             ('lte', '2021-06-02', 2),
             ('gt', '2021-06-02', 8),
             ('gte', '2021-06-02', 10),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-blockdata_date=%s&filter-blockdata_date-operator=%s'
                     % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         # email
         resp = get_app(pub).get(
             sign_uri('/api/forms/test/list?filter-blockdata_email=a@localhost', user=local_user)
-...
             sign_uri('/api/forms/test/list?filter-blockdata_email=c@localhost', user=local_user)
+        )
         assert len(resp.json) == 0
         params = [
             ('eq', 'a@localhost', 6),
             ('ne', 'a@localhost', 4),
+        ]
         for operator, value, result in params:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-blockdata_email=%s&filter-blockdata_email-operator=%s'
                     % (value, operator),
                     user=local_user,
+                )
+            )
             assert len(resp.json) == result
         for operator in ['lt', 'lte', 'gt', 'gte']:
             resp = get_app(pub).get(
                 sign_uri(
                     '/api/forms/test/list?filter-blockdata_email=plop0&filter-blockdata_email-operator=%s'
                     % operator,
                     user=local_user,
                 ),
                 status=400,
+            )
             assert resp.json['err_desc'] == 'Invalid operator "%s" for "filter-blockdata_email"' % operator
         # mix
         resp = get_app(pub).get(
             sign_uri(
                 '/api/forms/test/list?filter-blockdata_item=1&filter-blockdata_string=plop1', user=local_user
+            )
+        )
         assert len(resp.json) == 1
         assert len(resp.json) == 2
         resp = get_app(pub).get(
             sign_uri(
                 '/api/forms/test/list?filter-blockdata_item=2&filter-blockdata_string=plop1', user=local_user
+            )
+        )
         assert len(resp.json) == 0
         assert len(resp.json) == 1
         resp = get_app(pub).get(
             sign_uri(
                 '/api/forms/test/list?filter-blockdata_item=1&filter-blockdata_string=plop0', user=local_user

         Contains,
         Equal,
         FtsMatch,
         Greater,
         GreaterOrEqual,
         Intersects,
         Less,
         LessOrEqual,
         NotEqual,
         NotNull,
-...
             query_overrides = get_request().form
             return self.get_view_criterias(query_overrides, request=get_request())
         def get_view_criterias(self, query_overrides=None, request=None, record_errors=False):
             from wcs import sql
         def get_field_criteria(self, field, operator, field_key):
             mapping = {
                 'eq': Equal,
                 'ne': NotEqual,
                 'lt': Less,
                 'lte': LessOrEqual,
                 'gt': Greater,
                 'gte': GreaterOrEqual,
+            }
             types_with_all_ops = ['internal-id', 'date', 'item', 'items', 'string']
             if field.type not in types_with_all_ops and operator not in ['eq', 'ne']:
                 # eq and ne are always allowed
                 raise RequestError('Invalid operator "%s" for "%s"' % (operator, field_key))
             if operator not in mapping:
                 raise RequestError('Invalid operator "%s" for "%s"' % (operator, field_key))
             return mapping[operator]
         def get_view_criterias(self, query_overrides=None, request=None, record_errors=False):
             fake_fields = [
                 FakeField('internal-id', 'internal-id', _('Identifier')),
                 FakeField('number', 'number', _('Number')),
-...
             filters_in_request = {
                 k.replace('filter-', '')
                 for k in filters_dict
                 if k.startswith('filter-') and not k.endswith('-value')
                 if k.startswith('filter-') and not k.endswith('-value') and not k.endswith('-operator')
+            }
             filters_in_request = {
+                f
-...
                 if filter_field.type == 'user-id':
                     # convert uuid based filter into local id filter
                     name_id = filters_dict.get('filter-user-uuid')
                     filters_dict['filter-user-operator'] = filters_dict.get('filter-user-uuid-operator')
                     if name_id:
                         nameid_users = get_publisher().user_class.get_users_with_name_identifier(name_id)
                         request_form['filter-user'] = filters_dict['filter-user'] = 'on'
-...
                 if filter_field.type == 'submission-agent-id':
                     # convert uuid based filter into local id filter
                     name_id = filters_dict.get('filter-submission-agent-uuid')
                     filters_dict['filter-submission-agent-operator'] = filters_dict.get(
                         'filter-submission-agent-uuid-operator'
+                    )
                     if name_id:
                         nameid_users = get_publisher().user_class.get_users_with_name_identifier(name_id)
                         request_form['filter-submission-agent'] = filters_dict['filter-submission-agent'] = 'on'
-...
                 if not filter_field_value:
                     continue
                 if filter_field.type == 'date':
                 # get operator and criteria
                 filter_field_operator_key = '%s-operator' % filter_field_key.replace('-value', '')
                 filter_field_operator = filters_dict.get(filter_field_operator_key) or 'eq'
                 criteria = self.get_field_criteria(filter_field, filter_field_operator, filter_field_key)
                 # check value types
                 if filter_field_value is None and filter_field.type in [
                     'date',
                     'bool',
                     'item',
                     'items',
                     'string',
                     'email',
                 ]:
                     continue
                 if filter_field.type == 'internal-id':
                     try:
                         filter_field_value = int(filter_field_value)
                     except ValueError:
                         raise RequestError('Invalid value "%s" for "%s"' % (filter_field_value, filter_field_key))
                 elif filter_field.type == 'period-date':
                     try:
                         filter_date_value = misc.get_as_datetime(filter_field_value).timetuple()
                     except ValueError:
                         continue
                 elif filter_field.type == 'date':
                     try:
                         filter_field_value = misc.get_as_datetime(filter_field_value).date().strftime('%Y-%m-%d')
                     except ValueError:
-...
                         filter_field_value = False
                     else:
                         raise RequestError('Invalid value "%s" for "%s"' % (filter_field_value, filter_field_key))
                 elif filter_field.type in ('item', 'items', 'string'):
                     try:
                         filter_field_value = int(filter_field_value)
                     except ValueError:
                         pass
                 if getattr(filter_field, 'block_field', None):
                     criterias.append(
                         sql.ArrayContains(
                             'f%s' % filter_field.block_field.id,
                             json.dumps([{filter_field.id: filter_field_value}]),
                             parent_field=filter_field.block_field,
+                        )
+                    )
                 elif filter_field.type == 'internal-id':
                     criterias.append(Equal('id', str(filter_field_value)))
                 # add criteria
                 if filter_field.type == 'internal-id':
                     criterias.append(criteria('id', filter_field_value))
                 elif filter_field.type == 'number':
                     criterias.append(Equal('id_display', str(filter_field_value)))
                 elif filter_field.type == 'period-date':
                     try:
                         filter_date_value = misc.get_as_datetime(filter_field_value).timetuple()
                     except ValueError:
                         continue
                     if filter_field.id == 'start':
                         criterias.append(GreaterOrEqual('receipt_time', filter_date_value))
                     elif filter_field.id == 'end':
-...
                 elif filter_field.type == 'user-id':
                     if filter_field_value == '__current__' and get_request().user:
                         filter_field_value = str(get_request().user.id)
                     criterias.append(Equal('user_id', filter_field_value))
                     criterias.append(criteria('user_id', filter_field_value))
                 elif filter_field.type == 'submission-agent-id':
                     criterias.append(Equal('submission_agent_id', filter_field_value))
                 elif filter_field.type in ('item', 'items'):
                     if filter_field.type == 'item':
                         criterias.append(Equal('f%s' % filter_field.id, filter_field_value))
                     elif filter_field.type == 'items':
                         criterias.append(Intersects('f%s' % filter_field.id, [filter_field_value]))
                     field_options = filter_field.get_options()
                     if field_options and type(field_options[0]) in (list, tuple):
                         for option in field_options:
                             if filter_field_value in (option[0], option[-1]):
                                 filter_field_value = option[1]
                                 break
                     criterias[-1]._label = '%s: %s' % (filter_field.label, filter_field_value)
                 elif filter_field.type == 'bool':
                     criterias.append(Equal('f%s' % filter_field.id, filter_field_value))
                 elif filter_field.type in ('string', 'email'):
                     criterias.append(Equal('f%s' % filter_field.id, filter_field_value))
                 elif filter_field.type == 'date':
                     criterias.append(Equal('f%s' % filter_field.id, filter_field_value))
                     criterias.append(criteria('submission_agent_id', filter_field_value))
                 elif filter_field.type in ('item', 'items', 'bool', 'string', 'email', 'date'):
                     criterias.append(criteria('f%s' % filter_field.id, filter_field_value, field=filter_field))
                     if filter_field.type in ('item', 'items'):
                         field_options = filter_field.get_options()
                         if field_options and type(field_options[0]) in (list, tuple):
                             for option in field_options:
                                 if filter_field_value in (option[0], option[-1]):
                                     filter_field_value = option[1]
                                     break
                         criterias[-1]._label = '%s: %s' % (filter_field.label, filter_field_value)
             unknown_filters = sorted(filters_in_request - known_filters)
             if unknown_filters:

             self.field = kwargs.get('field')
         def build_lambda(self):
             return lambda x: self.op(getattr(x, self.attribute, None) or self.typed_none, self.value)
             def func(x):
                 if self.field and self.field.key in ['item', 'string'] and isinstance(self.value, int):
                     try:
                         _x = int(getattr(x, self.attribute, None))
                     except (ValueError, TypeError):
                         _x = self.typed_none
                     return self.op(_x, self.value)
                 if self.field and self.field.key == 'items':
                     _x = getattr(x, self.attribute, None) or []
                     if isinstance(self.value, int):
                         try:
                             _x = [int(e) for e in _x]
                         except ValueError:
                             _x = []
                     return getattr(self, 'array_op', any)(self.op(e, self.value) for e in _x)
                 if isinstance(self.value, int):
                     return self.op(int(getattr(x, self.attribute, None)) or self.typed_none, self.value)
                 return self.op(getattr(x, self.attribute, None) or self.typed_none, self.value)
             return func
         def __repr__(self):
             return '<%s (attribute: %r%s)>' % (
-...
     class NotEqual(Criteria):
         op = operator.ne
         array_op = all
     class LessOrEqual(Criteria):

             self.attribute = attribute.replace('-', '_')
             self.value = value
             self.field = kwargs.get('field')
             self.parent_field = kwargs.get('parent_field')
         def as_sql(self):
             if self.field and getattr(self.field, 'block_field', None):
                 # eq: EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' = 'value')
                 # lt: EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' < 'value')
                 # lte: EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' <= 'value')
                 # gt: EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' > 'value')
                 # gte: EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' >= 'value')
                 # with a NOT EXISTS and the opposite operator:
                 # ne: NOT EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' = 'value')
                 # note: aa->>'FOOBAR' can be written with an integer or bool cast
                 attribute = "aa->>'%s'" % self.field.id
                 if self.field.key in ['item', 'string'] and isinstance(self.value, int):
                     # integer cast of db values
                     attribute = "(CASE WHEN %s~E'^\\\\d+$' THEN (%s)::int ELSE NULL END)" % (attribute, attribute)
                 elif self.field.key == 'bool':
                     # bool cast of db values
                     attribute = '(%s)::bool' % attribute
                 return "%s(SELECT 1 FROM jsonb_array_elements(%s->'data') AS datas(aa) WHERE %s %s %%(c%s)s)" % (
                     getattr(self, 'sql_exists', 'EXISTS'),
                     get_field_id(self.field.block_field),
                     attribute,
                     getattr(self, 'sql_op_exists', self.sql_op),
                     id(self.value),
+                )
             attribute = self.attribute
             if self.field and self.field.key == 'items':
                 # eq: 'value' = ANY (ITEMS)
                 # ne: 'value' != ALL (ITEMS)
                 # with reversed operator:
                 # lt: 'value' > ANY (ITEMS)
                 # lte: 'value' >= ANY (ITEMS)
                 # gt: 'value' < ANY (ITEMS)
                 # gte: 'value' <= ANY (ITEMS)
                 # note: ITEMS is written with an integer cast or with a COALESCE expression
                 if isinstance(self.value, int):
                     # integer cast of db values
                     attribute = (
                         "CASE WHEN array_to_string(%s, '')~E'^\\\\d+$' THEN %s::int[] ELSE ARRAY[]::int[] END"
                         % (attribute, attribute)
+                    )
                 else:
                     # for none values
                     attribute = "COALESCE(%s, ARRAY[]::text[])" % attribute
                 return '%%(c%s)s %s %s (%s)' % (
                     id(self.value),
                     getattr(self, 'sql_reversed_op', self.sql_op),
                     getattr(self, 'sql_array_op', 'ANY'),
                     attribute,
+                )
             if self.field and self.field.key == 'computed':
                 attribute = "%s->>'data'" % self.attribute
             if self.parent_field and self.parent_field.key == 'block':
                 attribute = "%s->'data'" % self.attribute
             elif self.field and self.field.key in ['item', 'string'] and isinstance(self.value, int):
                 # integer cast of db values
                 attribute = "(CASE WHEN %s~E'^\\\\d+$' THEN %s::int ELSE NULL END)" % (attribute, attribute)
             return '%s %s %%(c%s)s' % (attribute, self.sql_op, id(self.value))
         def as_sql_param(self):
-...
     class Less(Criteria):
         sql_op = '<'
         sql_reversed_op = '>'
     class Greater(Criteria):
         sql_op = '>'
         sql_reversed_op = '<'
     class Equal(Criteria):
-...
     class LessOrEqual(Criteria):
         sql_op = '<='
         sql_reversed_op = '>='
     class GreaterOrEqual(Criteria):
         sql_op = '>='
         sql_reversed_op = '<='
     class NotEqual(Criteria):
         sql_op = '!='
         # in case of items field, we want to write this clause:
         # 'value' != ALL (ITEMS)
         sql_array_op = 'ALL'
         # in case of block field, we want to write this clause:
         # NOT EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' = 'value')
         # and not:
         # EXISTS (SELECT 1 FROM jsonb_array_elements(BLOCK->'data') AS datas(aa) WHERE aa->>'FOOBAR' != 'value')
         sql_exists = 'NOT EXISTS'
         sql_op_exists = '='
     class Contains(Criteria):
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-api-add-filter-operators-on-formdata-listing-60785.patch