Skip to content

erp5
erp5
Commit 025baa4c authored by Titouan Soulard's avatar Titouan Soulard
Browse files
Options

erp5_action_information_api: full support for OpenAPI

parent 2b887aea
Hide whitespace changes
Inline Side-by-side
Showing with 270 additions and 58 deletions
bt5/erp5_action_information_api/DocumentTemplateItem/portal_components/extension.erp5.OpenAPIHandler.py 0 → 100644
View file @ 025baa4c
import jsonschema
from six.moves.urllib.parse import unquote
from erp5.component.document.OpenAPITypeInformation import (
NoMethodForOperationError,
ParameterValidationError,
SchemaDefinitionError,
)
def getMatchingOperation(caller, request):
# type: (HTTPRequest) -> Optional[OpenAPIOperation]
# Compute the relative URL of the request path, by removing the
# relative URL of the Open API Service. This is tricky, because
# it may be in the acquisition context of a web section and the request
# might be using a virtual root with virtual paths.
# First, strip the left part of the URL corresponding to the "root"
web_section = caller.getWebSectionValue()
root = web_section if web_section is not None else caller.getPortalObject()
request_path_parts = [
unquote(part) for part in request['URL']
[1 + len(request.physicalPathToURL(root.getPhysicalPath())):].split('/')
]
# then strip everything corresponding to the "self" open api service.
# Here, unlike getPhysicalPath(), we don't use the inner acquistion,
# but keep the acquisition chain from this request traversal.
i = 0
for aq_parent in reversed(caller.aq_chain[:caller.aq_chain.index(root)]):
if aq_parent.id == request_path_parts[i]:
i += 1
else:
break
request_path_parts = request_path_parts[i:]
request_method = request.method.lower()
matched_operation = None
for operation in caller.getTypeInfo().getOpenAPIOperationIterator():
if operation.request_method != request_method:
continue
operation_path_parts = operation.path.split('/')[1:]
if len(operation_path_parts) != len(request_path_parts):
continue
if operation_path_parts == request_path_parts:
# this is a concrete match, use this operation
request.other['traverse_subpath'] = request_path_parts
return operation
# look for a templated match
for operation_path_part, request_path_part in zip(
operation_path_parts,
request_path_parts,
):
if operation_path_part == request_path_part:
continue
elif operation_path_part[0] == '{' and operation_path_part[-1] == '}':
continue
# TODO: match paths like /report.{format}
else:
break
else:
# we had a match, but there might be a "better" match, so we keep looping.
# https://spec.openapis.org/oas/v3.1.0.html#patterned-fields :
# > When matching URLs, concrete (non-templated) paths would be matched before
# > their templated counterparts
matched_operation = operation
continue
request.other['traverse_subpath'] = request_path_parts
return matched_operation
def getMethodForOperation(caller, operation):
# type: (OpenAPIOperation) -> Optional[Callable]
operation_id = operation.get('operationId')
if operation_id:
method = caller.getTypeBasedMethod(operation_id)
if method is not None:
return method
raise NoMethodForOperationError(
'No method for operation {operation_id} {request_method} {path}'.format(
operation_id=operation.get('operationId', ''),
request_method=operation.request_method.upper(),
path=operation.path,
))
def validateParameter(parameter_name, parameter_value, parameter, schema):
# type: (str, Any, dict, dict) -> Any
"""Validate the parameter (or request body), raising a ParameterValidationError
when the parameter is not valid according to the corresponding schema.
"""
if schema is not None:
if parameter_value is None and not parameter.get('required'):
return parameter_value
__traceback_info__ = (parameter_name, parameter_value, schema)
try:
jsonschema.validate(parameter_value, schema)
except jsonschema.ValidationError as e:
raise ParameterValidationError(
'Error validating {parameter_name}: {e}'.format(
parameter_name=parameter_name, e=e.message), str(e))
return parameter_value
def extractParametersFromRequest(operation, request):
# type: (OpenAPIOperation, HTTPRequest) -> dict
parameter_dict = {}
for parameter in operation.getParameters():
parameter_dict[parameter['name']] = validateParameter(
'parameter `{}`'.format(parameter['name']),
parameter.getValue(request),
parameter,
parameter.getJSONSchema(),
)
requestBody = validateParameter(
'request body',
operation.getRequestBodyValue(request),
{},
operation.getRequestBodyJSONSchema(request),
)
if requestBody:
# we try to bind the request body as `body` parameter, but use alternate name
# if it's already used by a parameter
for body_arg in ('body', 'request_body', 'body_'):
if body_arg not in parameter_dict:
parameter_dict[body_arg] = requestBody
break
else:
raise SchemaDefinitionError('unable to bind requestBody')
return parameter_dict
def returnDict(caller, portal, request):
schema = caller.getTypeInfo().getSchema()
schema.setdefault("servers", []).insert(
0, {
"url": caller.absolute_url(),
"description": caller.getDescription()
})
operation = getMatchingOperation(caller, request)
# XXX: unexpected behaviour, the schema should be returned only with the correct endpoint
if operation is None:
return schema
method = getMethodForOperation(caller, operation)
parameters = extractParametersFromRequest(operation, request)
return method(**parameters)
bt5/erp5_action_information_api/DocumentTemplateItem/portal_components/extension.erp5.OpenAPIHandler.xml 0 → 100644
View file @ 025baa4c
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Extension Component" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>default_reference</string> </key>
<value> <string>OpenAPIHandler</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>extension.erp5.OpenAPIHandler</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>text_content_error_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>version</string> </key>
<value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.Workflow"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_log</string> </key>
<value>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
bt5/erp5_action_information_api/SkinTemplateItem/portal_skins/erp5_action_information_api/ActionInformationAPI_api_openapi.py deleted 100644 → 0
View file @ 2b887aea
import json
schema = caller.getTypeInfo().getSchema()
schema.setdefault("servers", []).insert(
0, {
"url": caller.absolute_url(),
"description": caller.getDescription()
})
hyperdocument = json.dumps(schema)
return (hyperdocument, {})
bt5/erp5_action_information_api/SkinTemplateItem/portal_skins/erp5_action_information_api/ActionInformationAPI_api_openapi.xml
View file @ 025baa4c
......@@ -2,60 +2,26 @@
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
<global name="ExternalMethod" module="Products.ExternalMethod.ExternalMethod"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="_reconstructor" module="copy_reg"/>
</klass>
<tuple>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
<global name="object" module="__builtin__"/>
<none/>
</tuple>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
<key> <string>_function</string> </key>
<value> <string>returnDict</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>caller, portal, request</string> </value>
<key> <string>_module</string> </key>
<value> <string>OpenAPIHandler</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>ActionInformationAPI_api_openapi</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
......
bt5/erp5_action_information_api/TestTemplateItem/portal_components/test.erp5.testActionInformationAPI.py
View file @ 025baa4c
......@@ -169,14 +169,16 @@ return {"status": 200}""")
self.tic()
self.commit()
def loggedInRequest(self, path, method, content):
def loggedInRequest(self, path, method, content, env=None):
if env is None:
env = {"CONTENT_TYPE": "application/json"}
return self.publish(
self.web_service.getPath() + path,
request_method=method,
stdin=io.BytesIO(
json.dumps(content).encode()
),
env={"CONTENT_TYPE": "application/json"},
env=env,
user="ERP5TypeTestCase"
)
......@@ -220,3 +222,12 @@ return {"status": 200}""")
self.assertEqual(response.getBody(), json.dumps({ "status": 200 }, indent=2))
self.assertEqual(response.getStatus(), 200)
self.assertEqual(person.getDefaultEmailUrlString(), "alice@looking.glass")
def test_get_openapi(self):
self.setupOpenApi()
response = self.loggedInRequest("/pet/findByTags?tags=tag1", "GET", {}, env={})
body = json.loads(response.getBody())
self.assertEqual(response.getStatus(), 200)
self.assertEqual(len(body), 1)
bt5/erp5_action_information_api/bt/template_document_id_list
View file @ 025baa4c
document.erp5.ActionInformationAPI
\ No newline at end of file
document.erp5.ActionInformationAPI
extension.erp5.OpenAPIHandler
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7