Produits Entr'ouvert » Passerelle

            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

        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})

        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 {'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)

    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,

    )

    delete_view = 'opendatasoft-query-delete'

    edit_view = 'opendatasoft-query-edit'

    def q(self, request, **kwargs):

    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}}',

    )

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 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 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',