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 io

import operator

import zeep

import zeep.helpers

from django.db import models

from django.urls import reverse

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

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=True, verbose_name=_('Be strict with returned XML'))

    zeep_xsd_ignore_sequence_order = models.BooleanField(

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

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = _('SOAP connector')

    @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, **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

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

        serialized = zeep.helpers.serialize_object(soap_response)

        json_response = jsonify(serialized)

        return json_response

    @property

    def methods(self):

        def generator():

            for name in dir(self.client.service):

                if name.startswith('_'):

                    continue

                url = (

                    reverse(

                        'generic-endpoint',

                        kwargs={'connector': 'soap', 'endpoint': 'method', 'slug': self.slug},

                    )

                    + '/'

                    + name

                    + '/'

                )

                yield name, url

        return list(generator())

    @property

    def wsdl_dump(self):

        # get each operation signature

        dump = io.StringIO()

        def p(*args, **kwargs):

            print(*args, **kwargs, file=dump)

        for service in self.client.wsdl.services.values():

            p("service:", service.name)

            for port in service.ports.values():

                operations = sorted(port.binding._operations.values(), key=operator.attrgetter('name'))

                for operation in operations:

                    p("  method :", operation.name)

                    p("    input :", operation.input.signature())

                    p()

            p()

        return dump.getvalue()

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

{% load i18n passerelle %}

{% block endpoints %}

{{ block.super }}

<h4>{% trans "Available methods" %}</h4>

<ul class="endpoints">

    {% for name, url in object.methods %}

    <li class="get-method">

        <a href="{{ url }}">{{ site_base_uri }}{{ url }}</a>

    </li>

    {% endfor %}

</ul>

{% endblock %}

{% block extra-sections %}

<div id="soap-details" class="section">

  <h3>{% trans "WSDL dump" %}</h3>

  <pre>

  {{ object.wsdl_dump }}

  </pre>

</div>

{% endblock %}

    'passerelle.apps.soap',