0002-base-add-generic-BaseQuery-20535.patch

Valentin Deniaud, 06 avril 2020 18:28

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

Subject: [PATCH 2/5] base: add generic BaseQuery (#20535)

 passerelle/apps/arcgis/models.py              | 146 +++---------------
 .../templates/arcgis/arcgis_detail.html       |  16 --
 passerelle/base/models.py                     |  72 +++++++++
 tests/test_arcgis.py                          |  14 +-
 4 files changed, 108 insertions(+), 140 deletions(-)
 delete mode 100644 passerelle/apps/arcgis/templates/arcgis/arcgis_detail.html

     from passerelle.utils.api import endpoint
     from passerelle.utils.conversion import num2deg
     from passerelle.utils.templates import render_to_string, validate_template
     from passerelle.base.models import BaseResource, HTTPResource
     from passerelle.base.models import BaseResource, HTTPResource, BaseQuery
     class ArcGISError(APIError):
-...
                   description=_('Query'),
                   pattern=r'^(?P<query_slug>[\w:_-]+)/$',
                   perm='can_access',
                   parameters={
                       'q': {'description': _('Search text in display field')},
                       'full': {
                           'description': _('Returns all ArcGIS informations (geometry, metadata)'),
                           'type': 'bool',
                       },
                   },
                   show=False)
         def q(self, request, query_slug, q=None, full=False, **kwargs):
             query = get_object_or_404(Query, resource=self, slug=query_slug)
-...
             for key in request.GET:
                 if key not in refs:
                     kwargs[key] = request.GET[key]
             return query.q(request, q=None, full=full, **kwargs)
             return query.query(request, q=None, full=full, **kwargs)
         def export_json(self):
             d = super(ArcGIS, self).export_json()
-...
                 raise ValidationError(_('invalid reference'))
     @six.python_2_unicode_compatible
     class Query(models.Model):
     class Query(BaseQuery):
         resource = models.ForeignKey(
             to=ArcGIS,
             related_name='queries',
             verbose_name=_('Resource'))
         name = models.CharField(
             verbose_name=_('Name'),
             max_length=128)
         slug = models.SlugField(
             verbose_name=_('Slug'),
             max_length=128)
         description = models.TextField(
             verbose_name=_('Description'),
             blank=True)
         folder = models.CharField(
             verbose_name=_('ArcGis Folder'),
             max_length=64,
-...
             validators=[validate_template],
             blank=True)
         class Meta:
             unique_together = [
                 ('resource', 'name'),
                 ('resource', 'slug'),
+            ]
             ordering = ['name']
         delete_view = 'arcgis-query-delete'
         edit_view = 'arcgis-query-edit'
         @property
         def where_references(self):
-...
             else:
                 return []
         def q(self, request, q=None, full=False, **kwargs):
         def query(self, request, q=None, full=False, **kwargs):
             kwargs.update({
                 'service': self.service,
             })
-...
                 kwargs['where'] = formatter.format(self.where, **format_kwargs)
             return self.resource.mapservice_query(request, q=q, full=full, **kwargs)
         class QueryEndpoint:
             http_method = 'get'
             def __init__(self, query):
                 self.object = query.resource
                 self.query = query
                 self.name = 'q/%s/' % query.slug
             @property
             def description(self):
                 return self.query.name
             @property
             def long_description(self):
                 return self.query.description
             def example_url(self):
                 kwargs = {
                     'connector': self.object.get_connector_slug(),
                     'slug': self.object.slug,
                     'endpoint': self.name,
+                }
                 query_string = ''
                 if self.query.where_references:
                     query_string = '?' + '&'.join(
                         ['%s=%s' % (ref, '') for ref, _ in self.query.where_references + [('q', ''), ('full', '')]])
                 return reverse('generic-endpoint', kwargs=kwargs) + query_string
             def example_url_as_html(self):
                 kwargs = {
                     'connector': self.object.get_connector_slug(),
                     'slug': self.object.slug,
                     'endpoint': self.name,
+                }
                 query_string = ''
                 if self.query.where_references:
                     query_string = '?' + '&amp;'.join(
                         [format_html('{0}=<i class="varname">{1}</i>', ref, ref)
                          for ref, klass in self.query.where_references + [('q', ''), ('full', '')]])
                 return mark_safe(reverse('generic-endpoint', kwargs=kwargs) + query_string)
             def get_params(self):
                 params = []
                 for ref, klass in self.query.where_references:
                     params.append({
                         'name': ref,
                         'type': 'integer' if klass is int else 'string',
                     })
                 # Copy generic params descriptions from mapservice_query if they
                 # are not overloaded by the query
                 for param in self.object.mapservice_query.endpoint_info.get_params():
                     if (param['name'] in ('folder', 'service', 'layer', 'id_template')
                             and getattr(self.query, param['name'])):
                         continue
                     if param['name'] == 'template' and self.query.text_template:
                         continue
                     params.append(param)
                 return params
         @property
         def endpoint(self):
             return self.QueryEndpoint(self)
         def __str__(self):
             return self.name
         def export_json(self):
             d = {}
             fields = [
                 f for f in self.__class__._meta.get_fields()
                 if f.concrete and (
                     not f.is_relation
                     or f.one_to_one
                     or (f.many_to_one and f.related_model)
                 ) and f.name not in ['id', 'resource']
+            ]
             for field in fields:
                 d[field.name] = getattr(self, field.name)
             return d
         @classmethod
         def import_json(cls, d):
             return cls(**d)
         def delete_url(self):
             return reverse('arcgis-query-delete',
                            kwargs={'slug': self.resource.slug, 'pk': self.pk})
         def edit_url(self):
             return reverse('arcgis-query-edit',
                            kwargs={'slug': self.resource.slug, 'pk': self.pk})
         def as_endpoint(self):
             endpoint = super(Query, self).as_endpoint(path=self.resource.q.endpoint_info.name)
             # build parameters list from original mapservice endpoint
             mapservice_endpoint = self.resource.mapservice_query.endpoint_info
             endpoint.func = mapservice_endpoint.func
             endpoint.show_undocumented_params = False
             keep_params = ['folder', 'service', 'layer', 'id_template', 'q', 'full']
             if not self.text_template:
                 keep_params.append('template')
             endpoint.parameters = {name: mapservice_endpoint.parameters[name]
                                    for name in keep_params if not getattr(self, name, None)}
             if not self.where_references:
                 return endpoint
             for ref, klass in self.where_references:
                 endpoint.parameters[ref] = {'type': 'integer' if klass is int else 'string'}
             return endpoint

     {% extends "passerelle/manage/service_view.html" %}
     {% load i18n passerelle %}
     {% block endpoints %}
     {{ block.super }}
     {% if object.query_set.exists %}
     <h2>{% trans "Custom queries" %}</h2>
     <ul class="endpoints">
       {% for query in object.query_set.all %}
         {% include "passerelle/manage/endpoint.html" with endpoint=query.endpoint %}
       {% endfor %}
     </ul>
     {% endif %}
     {% endblock %}

                         endpoint_info.http_method = http_method
                         endpoints.append(endpoint_info)
             endpoints.sort(key=lambda x: (x.name or '', x.pattern or ''))
             if hasattr(self, 'queries'):
                 self.append_custom_queries(endpoints)
             return endpoints
         def append_custom_queries(self, endpoints):
             for query in self.queries.all():
                 if hasattr(query, 'as_endpoint'):
                     endpoints.append(query.as_endpoint())
         def get_connector_permissions(self):
             perms = {}
             for endpoint_info in self.get_endpoints_infos():
-...
         def __str__(self):
             return '%s %s %s' % (self.timestamp, self.appname, self.slug)
     @six.python_2_unicode_compatible
     class BaseQuery(models.Model):
         '''Base for building custom queries.
         It must define "resource" attribute as a ForeignKey to a BaseResource subclass,
         and probably extend its "as_endpoint" method to document its parameters.
         '''
         name = models.CharField(
             verbose_name=_('Name'),
             max_length=128)
         slug = models.SlugField(
             verbose_name=_('Slug'),
             max_length=128)
         description = models.TextField(
             verbose_name=_('Description'),
             blank=True)
         http_method = 'get'
         class Meta:
             abstract = True
             unique_together = [
                 ('resource', 'name'),
                 ('resource', 'slug'),
+            ]
             ordering = ['name']
         def as_endpoint(self, path=''):
             name = '%s/%s/' % (path, self.slug) if path else self.slug + '/'
             e = endpoint(name=name, description=self.name, long_description=self.description)
             e.http_method = self.http_method
             e.object = self.resource
             return e
         def __str__(self):
             return self.name
         def export_json(self):
             d = {}
             fields = [
                 f for f in self.__class__._meta.get_fields()
                 if f.concrete and (
                     not f.is_relation
                     or f.one_to_one
                     or (f.many_to_one and f.related_model)
                 ) and f.name not in ['id', 'resource']
+            ]
             for field in fields:
                 d[field.name] = getattr(self, field.name)
             return d
         @classmethod
         def import_json(cls, d):
             return cls(**d)
         def delete_url(self):
             return reverse(self.delete_view,
                            kwargs={'slug': self.resource.slug, 'pk': self.pk})
         def edit_url(self):
             return reverse(self.edit_view,
                            kwargs={'slug': self.resource.slug, 'pk': self.pk})

             resource=arcgis,
             name='Adresses',
             slug='adresses',
             description='Recherche d\'une adresse',
             description='Rechercher une adresse',
             id_template='{{ attributes.ident }}',
             text_template='{{ attributes.address }} - {{ attributes.codepost }}',
             folder='fold',
-...
             where='adress LIKE {adress:s}')
     def test_query_q_method(arcgis, query, rf):
     def test_query_query_method(arcgis, query, rf):
         with mock.patch('passerelle.utils.Request.get') as requests_get:
             requests_get.return_value = utils.FakedResponse(content=STATES,
                                                             status_code=200)
             query.q(rf.get('/', data={'adress': "AVENUE D'ANNAM"}), full=True) == []
             query.query(rf.get('/', data={'adress': "AVENUE D'ANNAM"}), full=True) == []
             assert requests_get.call_count == 1
             assert requests_get.call_args[0][0] == 'https://arcgis.example.net/services/fold/serv/MapServer/1/query'
             args = requests_get.call_args[1]['params']
-...
             assert resp.content_type == 'image/png'
     def test_query_documentation(arcgis, query, app):
         resp = app.get(arcgis.get_absolute_url())
         assert query.name in resp.text
         assert query.description in resp.text
         # additional parameter appears in endpoint documentation
         assert '<span class="param-name">adress</span>' in resp.text
     def test_export_import(query):
         assert ArcGIS.objects.count() == 1
         assert Query.objects.count() == 1
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0002-base-add-generic-BaseQuery-20535.patch