0003-opendatasoft-add-facet-filters-50212.patch

Nicolas Roche, 25 juin 2021 19:33

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

Subject: [PATCH 3/3] opendatasoft: add facet filters (#50212)

 .../migrations/0002_auto_20210625_1852.py     | 33 +++++++++++++
 passerelle/apps/opendatasoft/models.py        | 48 +++++++++++++------
 tests/test_opendatasoft.py                    | 20 ++++++++
 3 files changed, 87 insertions(+), 14 deletions(-)
 create mode 100644 passerelle/apps/opendatasoft/migrations/0002_auto_20210625_1852.py

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.29 on 2021-06-25 16:52
     from __future__ import unicode_literals
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('opendatasoft', '0001_initial'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='query',
                 name='filter_expression',
                 field=models.TextField(
                     blank=True,
                     help_text='Specify refine and exclude facet expressions separated lines',
                     verbose_name='filter',
                 ),
             ),
             migrations.AlterField(
                 model_name='opendatasoft',
                 name='service_url',
                 field=models.CharField(
                     help_text='URL without ending "api/records/1.0/search/"',
                     max_length=256,
                     verbose_name='Site URL',
                 ),
             ),
+        ]

                 Query.objects.filter(resource=instance).delete()
             for data_query in data_queries:
                 query = Query.import_json(data_query)
                 query.resource = instance
                 queries.append(query)
             Query.objects.bulk_create(queries)
             return instance
         @endpoint(
             perm='can_access',
             description=_('Search'),
             parameters={
                 'dataset': {'description': _('Dataset')},
                 'text_template': {'description': _('Text template')},
                 'id': {'description': _('Record identifier')},
                 'q': {'description': _('Full text query')},
                 'limit': {'description': _('Maximum items')},
             },
+        )
         def search(self, request, dataset=None, text_template='', id=None, q=None, limit=None, **kwargs):
         def call_search(self, dataset=None, text_template='', filter_expression='', id=None, q=None, limit=None):
             scheme, netloc, path, params, query, fragment = urlparse.urlparse(self.service_url)
             path = urlparse.urljoin(path, 'api/records/1.0/search/')
             url = urlparse.urlunparse((scheme, netloc, path, params, query, fragment))
             if id is not None:
                 query = 'recordid:%s' % id
             else:
                 query = q
             params = {
                 'dataset': dataset,
                 'q': query,
+            }
             if self.api_key:
                 params.update({'apikey': self.api_key})
             if limit:
                 params.update({'rows': limit})
             params.update(urlparse.parse_qs(filter_expression))
             result_response = self.requests.get(url, params=params)
             err_desc = result_response.json().get('error')
             if err_desc:
                 raise APIError(err_desc, http_status=200)
             result = []
             for record in result_response.json().get('records'):
                 data = {}
                 for key, value in record.get('fields').items():
                     data[key] = value
                 data['id'] = record.get('recordid')
                 data['text'] = render_to_string(text_template, data).strip()
                 result.append(data)
             return result
         @endpoint(
             perm='can_access',
             description=_('Search'),
             parameters={
                 'dataset': {'description': _('Dataset')},
                 'text_template': {'description': _('Text template')},
                 'id': {'description': _('Record identifier')},
                 'q': {'description': _('Full text query')},
                 'limit': {'description': _('Maximum items')},
             },
+        )
         def search(self, request, dataset=None, text_template='', id=None, q=None, limit=None, **kwargs):
             result = self.call_search(dataset, text_template, '', id, q, limit)
             return {'data': result}
         @endpoint(
             name='q',
             description=_('Query'),
             pattern=r'^(?P<query_slug>[\w:_-]+)/$',
             perm='can_access',
             show=False,
+        )
         def q(self, request, query_slug, **kwargs):
             query = get_object_or_404(Query, resource=self, slug=query_slug)
             return query.q(request, **kwargs)
             result = query.q(request, **kwargs)
             return {'data': result}
         def create_query_url(self):
             return reverse('opendatasoft-query-new', kwargs={'slug': self.slug})
     class Query(BaseQuery):
         resource = models.ForeignKey(
             to=OpenDataSoft, related_name='queries', verbose_name=_('Resource'), on_delete=models.CASCADE
-...
             help_text=_('dataset to query'),
+        )
         text_template = models.TextField(
             verbose_name=_('Text template'),
             help_text=_("Use Django's template syntax. Attributes can be accessed through {{ attributes.name }}"),
             validators=[validate_template],
             blank=True,
+        )
         filter_expression = models.TextField(
             verbose_name=_('filter'),
             help_text=_('Specify refine and exclude facet expressions separated lines'),
             blank=True,
+        )
         delete_view = 'opendatasoft-query-delete'
         edit_view = 'opendatasoft-query-edit'
         def q(self, request, **kwargs):
             return self.resource.search(request, dataset=self.dataset, text_template=self.text_template, **kwargs)
             return self.resource.call_search(
                 dataset=self.dataset,
                 text_template=self.text_template,
                 filter_expression='&'.join(
                     [x.strip() for x in str(self.filter_expression).splitlines() if x.strip()]
                 ),
                 id=kwargs.get('id'),
                 q=kwargs.get('q'),
                 limit=kwargs.get('limit'),
+            )
         def as_endpoint(self):
             endpoint = super(Query, self).as_endpoint(path=self.resource.q.endpoint_info.name)
             search_endpoint = self.resource.search.endpoint_info
             endpoint.func = search_endpoint.func
             endpoint.show_undocumented_params = False

     def query(connector):
         return Query.objects.create(
             resource=connector,
             name='Référenciel adresses de test',
             slug='my_query',
             description='Rechercher une adresse',
             dataset='referentiel-adresse-test',
             text_template='{{numero}} {{nom_rue}} {{nom_commun}}',
             filter_expression='''
     refine.source=Ville et Eurométropole de Strasbourg
     exclude.numero=42
     exclude.numero=43
     ''',
+        )
     def test_views(db, admin_user, app, connector):
         app = login(app)
         resp = app.get('/opendatasoft/my_connector/', status=200)
         resp = resp.click('New Query')
         resp.form['name'] = 'my query'
-...
     def test_query_q_using_q(mocked_get, app, query):
         endpoint = '/opendatasoft/my_connector/q/my_query/'
         params = {
             'q': "rue de l'aubepine",
             'limit': 3,
+        }
         mocked_get.return_value = utils.FakedResponse(content=FAKED_CONTENT_Q_SEARCH, status_code=200)
         resp = app.get(endpoint, params=params, status=200)
         assert mocked_get.call_args[1]['params'] == {
             'dataset': 'referentiel-adresse-test',
             'q': "rue de l'aubepine",
             'apikey': 'my_secret',
             'rows': '3',
             'refine.source': ['Ville et Eurométropole de Strasbourg'],
             'exclude.numero': ['42', '43'],
+        }
         assert not resp.json['err']
         assert len(resp.json['data']) == 3
         # check order is kept
         assert [x['id'] for x in resp.json['data']] == [
             'e00cf6161e52a4c8fe510b2b74d4952036cb3473',
             '7cafcd5c692773e8b863587b2d38d6be82e023d8',
             '0984a5e1745701f71c91af73ce764e1f7132e0ff',
+        ]
-...
     @mock.patch('passerelle.utils.Request.get')
     def test_query_q_using_id(mocked_get, app, query):
         endpoint = '/opendatasoft/my_connector/q/my_query/'
         params = {
             'id': '7cafcd5c692773e8b863587b2d38d6be82e023d8',
+        }
         mocked_get.return_value = utils.FakedResponse(content=FAKED_CONTENT_ID_SEARCH, status_code=200)
         resp = app.get(endpoint, params=params, status=200)
         assert mocked_get.call_args[1]['params'] == {
             'dataset': 'referentiel-adresse-test',
             'q': 'recordid:7cafcd5c692773e8b863587b2d38d6be82e023d8',
             'apikey': 'my_secret',
             'refine.source': ['Ville et Eurométropole de Strasbourg'],
             'exclude.numero': ['42', '43'],
+        }
         assert len(resp.json['data']) == 1
         assert resp.json['data'][0]['text'] == "19 RUE DE L'AUBEPINE Lipsheim"
     def test_opendatasoft_query_unicity(admin_user, app, connector, query):
         connector2 = OpenDataSoft.objects.create(
             slug='my_connector2',
             api_key='my_secret',
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0003-opendatasoft-add-facet-filters-50212.patch