passerelle/passerelle/apps/matrix42/models.py

# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2023 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/>.


from urllib.parse import urljoin

from django.db import models
from django.utils.translation import gettext_lazy as _

from passerelle.base.models import BaseResource, HTTPResource
from passerelle.utils.api import endpoint
from passerelle.utils.jsonresponse import APIError
from passerelle.utils.templates import render_to_string

DICT_SCHEMA = {
    '$schema': 'http://json-schema.org/draft-04/schema#',
    'type': 'object',
    'additionalProperties': True,
    'unflatten': True,
}


class Matrix42(BaseResource, HTTPResource):
    category = _('Business Process Connectors')

    log_requests_errors = False

    class Meta:
        verbose_name = _('Matrix42 Public API')

    base_url = models.URLField(
        _('Webservice Base URL'), help_text=_('Example: https://xxx.m42cloud.com/m42Services/api/')
    )
    token = models.CharField(max_length=512, verbose_name=_('Authorization Token'))

    def get_authorization_headers(self):
        headers = {
            'Content-Type': 'application/json',
            'Authorization': 'Bearer ' + self.token,
        }
        token = self.request('ApiToken/GenerateAccessTokenFromApiToken', headers=headers, method='POST')
        if 'RawToken' not in token:
            raise APIError('Matrix42 not returned a RawToken: %s' % token)
        return {
            'Content-Type': 'application/json',
            'Authorization': 'Bearer ' + token['RawToken'],
        }

    def request(self, uri, params=None, json=None, headers=None, method=None, dict_response=True):
        if headers is None:
            headers = self.get_authorization_headers()
        if method is None:
            method = 'GET' if json is None else 'POST'
        url = urljoin(self.base_url, uri)
        response = self.requests.request(method, url, params=params, json=json, headers=headers)
        status_code = response.status_code
        try:
            response = response.json()
        except ValueError:
            raise APIError(
                'Matrix42 returned %s response with invalid JSON content: %r'
                % (status_code, response.content)
            )
        if dict_response:
            if not isinstance(response, dict):
                raise APIError(
                    'Matrix42 returned %s response, not returned a dict: %r' % (status_code, response),
                    data=response,
                )
        if isinstance(response, dict) and 'ExceptionName' in response:
            message = response.get('Message') or '(no message)'
            raise APIError(
                'Matrix42 returned %s response, ExceptionName "%s": %s'
                % (status_code, response['ExceptionName'], message),
                data=response,
            )
        if status_code // 100 != 2:
            raise APIError('Matrix42 returned status code %s' % status_code, data=response)
        return response

    @endpoint(
        name='fragment',
        description=_('Fragment Query'),
        display_category=_('Fragments'),
        parameters={
            'ddname': {
                'description': _('Technical name of the Data Definition'),
                'example_value': 'SPSUserClassBase',
            },
            'columns': {
                'description': _('Columns in the result set, separated by comma'),
                'example_value': 'ID,[Expression-ObjectID] as EOID,LastName,FirstName,MailAddress',
            },
            'template': {
                'description': _(
                    'Django template for text attribute - if none, use DisplayString|DisplayName|Name'
                ),
                'example_value': '{{ FirstName }} {{ LastName }} ({{ MailAddress }})',
            },
            'id_template': {
                'description': _('Django template for id attribute - if none, use ID'),
                'example_value': '{{ ID }}',
            },
            'search_column': {
                'description': _('Column for "q" search'),
            },
            'q': {'description': _('Search text in search column')},
            'id': {'description': _('Get the whole fragment with this ID')},
        },
    )
    def fragment(
        self,
        request,
        ddname,
        columns=None,
        template=None,
        id_template=None,
        search_column=None,
        q=None,
        id=None,
    ):
        def add_id_and_text(result):
            if id_template:
                result['id'] = render_to_string(id_template, result)
            else:
                result['id'] = result.get('ID')
            if template:
                result['text'] = render_to_string(template, result)
            else:
                result['text'] = (
                    result.get('DisplayString') or result.get('DisplayName') or result.get('Name') or ''
                )

        if id:
            uri = 'data/fragments/%s/%s' % (ddname, id)
            result = self.request(uri)
            add_id_and_text(result)
            return {'data': [result]}

        if q is not None and not search_column:
            raise APIError('q needs a search_column parameter', http_status=400)

        uri = urljoin(self.base_url, 'data/fragments/%s/schema-info' % ddname)
        params = {}
        if columns:
            params['columns'] = columns
        if q is not None:
            params['where'] = "%s LIKE '%%%s%%'" % (search_column, q.replace("'", "''"))
        results = self.request(uri, params=params).get('Result') or []
        for result in results:
            add_id_and_text(result)
        return {'data': results}

    @endpoint(
        name='get-object',
        description=_('Get an object'),
        display_category=_('Objects'),
        methods=['get'],
        pattern=r'^(?P<ciname>.+)/(?P<object_id>.+)$',
        example_pattern='SPSActivityTypeTicket/01b02f7d-adb6-49e6-aae3-66251ecbf98e',
    )
    def get_object(
        self,
        request,
        ciname,
        object_id,
    ):
        uri = urljoin(self.base_url, 'data/objects/%s/%s' % (ciname, object_id))
        return {'data': self.request(uri)}

    @endpoint(
        name='create-object',
        display_category=_('Objects'),
        methods=['post'],
        pattern=r'^(?P<ciname>.+)$',
        example_pattern='SPSActivityTypeTicket',
        post={
            'description': _('Create an new object'),
            'request_body': {'schema': {'application/json': DICT_SCHEMA}},
        },
    )
    def create_object(
        self,
        request,
        ciname,
        post_data,
    ):
        uri = urljoin(self.base_url, 'data/objects/%s' % ciname)
        return {'data': self.request(uri, json=post_data, dict_response=False)}