From an OpenAPI definition file I generated a Swagger client for Python using Swagger-UI. I'm having trouble using a POST function that requires a request body. What would be the correct approach here?
The definition defines the following:
"/xyz" : {
"post" : {
"tags" : [ "xyz" ],
"operationId" : "xyz",
"requestBody" : {
"content" : {
"multipart/form-data" : {
"schema" : {
"required" : [ "d1", "d2" ],
"type" : "object",
"properties" : {
"d1" : {
"format" : "binary",
"type" : "string"
},
"d2" : {
"format" : "email",
"type" : "string",
"nullable" : false
}
}
}
}
},
"required" : true
},
The method exists:
my_api.xyz(...)
But I dont know how to use the paramters. I found out, the the method needs a body parameter:
my_api.xyz(body)
d1 have to be the content of a xml-file.
I pasted your example schema into https://editor-next.swagger.io/ and edited enough to generate a Python client, by adding the wrapper object like:
{
"openapi": "3.0.3",
"paths": {
"/xyz": {
"post": {
"tags": [ "xyz" ],
"operationId": "xyz",
"requestBody": {
"content": {
"multipart/form-data": {
"schema": {
"required": [ "d1", "d2" ],
"type": "object",
"properties": {
"d1": {
"format": "binary",
"type": "string"
},
"d2": {
"format": "email",
"type": "string",
"nullable": false
}
}
}
}
},
"required" : true
}
}
}
}
}
In the zip of files that it generates is swagger_client/api/xyz_api.py
from __future__ import absolute_import
import re # noqa: F401
# python 2 and python 3 compatibility library
import six
from swagger_client.api_client import ApiClient
class XyzApi(object):
"""NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
Ref: https://github.com/swagger-api/swagger-codegen
"""
def __init__(self, api_client=None):
if api_client is None:
api_client = ApiClient()
self.api_client = api_client
def xyz(self, d1, d2, **kwargs): # noqa: E501
"""xyz # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.xyz(d1, d2, async_req=True)
>>> result = thread.get()
:param async_req bool
:param str d1: (required)
:param str d2: (required)
:return: None
If the method is called asynchronously,
returns the request thread.
"""
kwargs['_return_http_data_only'] = True
if kwargs.get('async_req'):
return self.xyz_with_http_info(d1, d2, **kwargs) # noqa: E501
else:
(data) = self.xyz_with_http_info(d1, d2, **kwargs) # noqa: E501
return data
def xyz_with_http_info(self, d1, d2, **kwargs): # noqa: E501
"""xyz # noqa: E501
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please pass async_req=True
>>> thread = api.xyz_with_http_info(d1, d2, async_req=True)
>>> result = thread.get()
:param async_req bool
:param str d1: (required)
:param str d2: (required)
:return: None
If the method is called asynchronously,
returns the request thread.
"""
all_params = ['d1', 'd2'] # noqa: E501
all_params.append('async_req')
all_params.append('_return_http_data_only')
all_params.append('_preload_content')
all_params.append('_request_timeout')
params = locals()
for key, val in six.iteritems(params['kwargs']):
if key not in all_params:
raise TypeError(
"Got an unexpected keyword argument '%s'"
" to method xyz" % key
)
params[key] = val
del params['kwargs']
# verify the required parameter 'd1' is set
if ('d1' not in params or
params['d1'] is None):
raise ValueError("Missing the required parameter `d1` when calling `xyz`") # noqa: E501
# verify the required parameter 'd2' is set
if ('d2' not in params or
params['d2'] is None):
raise ValueError("Missing the required parameter `d2` when calling `xyz`") # noqa: E501
collection_formats = {}
path_params = {}
query_params = []
header_params = {}
form_params = []
local_var_files = {}
if 'd1' in params:
local_var_files['d1'] = params['d1'] # noqa: E501
if 'd2' in params:
form_params.append(('d2', params['d2'])) # noqa: E501
body_params = None
# HTTP header `Content-Type`
header_params['Content-Type'] = self.api_client.select_header_content_type( # noqa: E501
['multipart/form-data']) # noqa: E501
# Authentication setting
auth_settings = [] # noqa: E501
return self.api_client.call_api(
'/xyz', 'POST',
path_params,
query_params,
header_params,
body=body_params,
post_params=form_params,
files=local_var_files,
response_type=None, # noqa: E501
auth_settings=auth_settings,
async_req=params.get('async_req'),
_return_http_data_only=params.get('_return_http_data_only'),
_preload_content=params.get('_preload_content', True),
_request_timeout=params.get('_request_timeout'),
collection_formats=collection_formats)
We can see that the d1 param is treated as a file, as we'd expect for "format" : "binary" in the schema
If we look in the generic api client implementation at swagger_client/api_client.py we can see how file params are handled:
def __call_api(
self, resource_path, method, path_params=None,
query_params=None, header_params=None, body=None, post_params=None,
files=None, response_type=None, auth_settings=None,
_return_http_data_only=None, collection_formats=None,
_preload_content=True, _request_timeout=None):
...
# post parameters
if post_params or files:
post_params = self.prepare_post_parameters(post_params, files)
and:
def prepare_post_parameters(self, post_params=None, files=None):
"""Builds form parameters.
:param post_params: Normal form parameters.
:param files: File parameters.
:return: Form parameters with files.
"""
params = []
if post_params:
params = post_params
if files:
for k, v in six.iteritems(files):
if not v:
continue
file_names = v if type(v) is list else [v]
for n in file_names:
with open(n, 'rb') as f:
filename = os.path.basename(f.name)
filedata = f.read()
mimetype = (mimetypes.guess_type(filename)[0] or
'application/octet-stream')
params.append(
tuple([k, tuple([filename, filedata, mimetype])]))
return params
So it appears to expect the d1 param value to be a filename (or list of filenames), i.e. path to a file on the local file system where the client is running, it then reads the content of the file and sends it in the POST body of the request.