Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.24 on 2022-01-19 16:37

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='SOAPConnector',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                (

                    'basic_auth_username',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication username'

                    ),

                ),

                (

                    'basic_auth_password',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication password'

                    ),

                ),

                (

                    'client_certificate',

                    models.FileField(

                        blank=True, null=True, upload_to='', verbose_name='TLS client certificate'

                    ),

                ),

                (

                    'trusted_certificate_authorities',

                    models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),

                ),

                (

                    'verify_cert',

                    models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'),

                ),

                (

                    'http_proxy',

                    models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),

                ),

                (

                    'wsdl_url',

                    models.URLField(

                        help_text='URL of the WSDL file', max_length=400, verbose_name='WSDL URL'

                    ),

                ),

                (

                    'zeep_strict',

                    models.BooleanField(default=True, verbose_name='Be strict with returned XML'),

                ),

                (

                    'zeep_xsd_ignore_sequence_order',

                    models.BooleanField(default=False, verbose_name='Ignore sequence order'),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_soapconnector_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'SOAP connector',

            },

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2019  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import collections

import zeep

import zeep.helpers

import zeep.xsd

from django.db import models

from django.utils.functional import cached_property

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource, HTTPResource

from passerelle.utils.api import endpoint

from passerelle.utils.json import unflatten

class SOAPConnector(BaseResource, HTTPResource):

    wsdl_url = models.URLField(

        max_length=400, verbose_name=_('WSDL URL'), help_text=_('URL of the WSDL file')

    )

    zeep_strict = models.BooleanField(default=False, verbose_name=_('Be strict with returned XML'))

    zeep_xsd_ignore_sequence_order = models.BooleanField(

        default=True, verbose_name=_('Ignore sequence order')

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('SOAP connector')

    @classmethod

    def get_manager_form_class(cls, **kwargs):

        form_class = super().get_manager_form_class(**kwargs)

        fields = list(form_class.base_fields.items())

        form_class.base_fields = collections.OrderedDict(fields[:2] + fields[-3:] + fields[2:-3])

        return form_class

    @cached_property

    def client(self):

        return self.soap_client(

            wsdl_url=self.wsdl_url,

            settings=zeep.Settings(

                strict=self.zeep_strict, xsd_ignore_sequence_order=self.zeep_xsd_ignore_sequence_order

            ),

        )

    @endpoint(

        methods=['get', 'post'],

        perm='can_access',

        name='method',

        pattern=r'^(?P<method_name>\w+)/$',

        example_pattern='method_name/',

        description=_('Call a SOAP method'),

    )

    def method(self, request, method_name, post_data=None, **kwargs):

        def jsonify(data):

            if isinstance(data, (dict, collections.OrderedDict)):

                # ignore _raw_elements, zeep put there nodes not maching the

                # XSD when strict parsing is disabled.

                return {

                    jsonify(k): jsonify(v)

                    for k, v in data.items()

                    if (self.zeep_strict or k != '_raw_elements')

                }

            elif isinstance(data, (list, tuple, collections.deque)):

                return [jsonify(item) for item in data]

            else:

                return data

        payload = {k: request.getlist(k) for k in request.GET if k != 'raise'}

        payload.update(unflatten(post_data or {}))

        soap_response = getattr(self.client.service, method_name)(**payload)

        serialized = zeep.helpers.serialize_object(soap_response)

        json_response = jsonify(serialized)

        return {'err': 0, 'data': json_response}

    def get_endpoints_infos(self):

        endpoints = super().get_endpoints_infos()

        for name, input_schema, output_schema in self.operations_and_schemas:

            kwargs = dict(

                name='method',

                pattern=f'{name}/',

                example_pattern=f'{name}/',

                description=f'Method {name}',

                json_schema_response={

                    'type': 'object',

                    'properties': collections.OrderedDict(

                        [

                            ('err', {'type': 'integer'}),

                            ('data', output_schema),

                        ]

                    ),

                },

            )

            if input_schema:

                kwargs['post_json_schema'] = input_schema

            endpoints.append(endpoint(**kwargs))

            endpoints[-1].object = self

            endpoints[-1].func = self.method

            if input_schema:

                endpoints[-1].http_method = 'post'

            else:

                endpoints[-1].http_method = 'get'

        return endpoints

    @property

    def operations_and_schemas(self):

        operations = self.client.service._binding._operations

        for name in operations:

            operation = operations[name]

            input_type = operation.input.body.type

            output_type = operation.output.body.type

            input_schema = self.type2schema(input_type, root=True)

            output_schema = self.type2schema(output_type, root=True)

            yield name, input_schema, output_schema

    def type2schema(self, xsd_type, root=False):

        # simplify schema: when a type contains a unique element, it will try

        # to match any dict or list with it on input and will flatten the

        # schema on output.

        if isinstance(xsd_type, zeep.xsd.ComplexType) and len(xsd_type.elements) == 1 and not root:

            if xsd_type.elements[0][1].max_occurs != 1:

                return {'type': 'array', 'items': self.type2schema(xsd_type.elements[0][1].type)}

            return self.type2schema(xsd_type.elements[0][1].type)

        if isinstance(xsd_type, zeep.xsd.ComplexType):

            properties = collections.OrderedDict()

            schema = {

                'type': 'object',

                'properties': properties,

            }

            for key, element in xsd_type.elements:

                if element.min_occurs > 0:

                    schema.setdefault('required', []).append(key)

                element_schema = self.type2schema(element.type)

                if element.max_occurs == 'unbounded' or element.max_occurs > 1:

                    element_schema = {'type': 'array', 'items': element_schema}

                properties[key] = element_schema

            if not properties:

                return None

            return schema

        if isinstance(xsd_type, zeep.xsd.BuiltinType):

            return {'type': 'string'}

        return f'!!! UNKNOWN TYPE {xsd_type} !!!'

{% extends "passerelle/manage/service_view.html" %}

{% load i18n passerelle %}

    'passerelle.apps.soap',