如何使用python swagger客户端发送二进制字符串

6rvt4ljy  于 2023-08-05  发布在  Python
关注(0)|答案(1)|浏览(105)

从OpenAPI定义文件中,我使用Swagger-UI为Python生成了一个Swagger客户端。我在使用需要请求主体的POST函数时遇到问题。在此,正确的方法是什么?
该定义定义了以下内容:

"/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
    },

字符串
该方法存在于:

my_api.xyz(...)


但是我不知道如何使用这些参数。我发现,该方法需要一个body参数:

my_api.xyz(body)


d1必须是xml文件的内容。

wf82jlnq

wf82jlnq1#

我将你的示例模式粘贴到https://editor-next.swagger.io/中,并通过添加 Package 器对象来编辑足够的内容以生成Python客户端,如:

{
  "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
        }
      }
    }
  }
}

字符串
在它生成的文件的zip中是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)


我们可以看到,d1参数被视为一个文件,正如我们在模式中对"format" : "binary"所期望的那样
如果我们查看swagger_client/api_client.py上的通用API客户端实现,我们可以看到文件参数是如何处理的:

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)


以及:

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


因此,似乎期望d1参数值是文件名(或文件名列表),即路径到客户端正在运行的本地文件系统上的文件,然后读取文件的内容并在请求的POST主体中发送。

相关问题