209 lines
7.6 KiB
Python
209 lines
7.6 KiB
Python
# 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': _('Search column: "where search_column LIKE \'%q%\'"'),
|
|
},
|
|
'search_filter': {
|
|
'description': _('Search filter: "where search_column LIKE \'%q%\' and search_filter"'),
|
|
},
|
|
'q': {'description': _('Search text (needs a 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,
|
|
search_filter=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("'", "''"))
|
|
if search_filter:
|
|
params['where'] += ' AND %s' % search_filter
|
|
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)}
|