string($binary)

Question

Помогите пожалуйста разобраться как передать бинарное значение файла 'string' через Python следуя данному body, указанному в документации Api :

    {
      "file": "string",
      "data": {
        "force_update": false,
        "only_available": false,
        "mark_missing_product_as": "none",
        "updated_fields": [
          "price",
          "presence"
        ]
      }
    }

Так же в документации указано что значение "string" является file string($binary) - xls,csv,xlsx,xml,cml документ. Со всем кодом разобрался , кроме этого момента где нужно передать string. Вот мой код на данный момент:

import json
import http.client
import pprint


# API Settigs
AUTH_TOKEN = 'token'  # Your authorization token
HOST = 'my.prom.ua'  # e.g.: my.prom.ua, my.tiu.ru, my.satu.kz, my.deal.by, my.prom.md


class HTTPSError(Exception):
    pass


class EvoClientExample(object):

    def __init__(self, token):
        self.token = token

    def make_request(self, method, url, body=None):
        connection = http.client.HTTPSConnection(HOST)

        headers = {'Authorization': 'Bearer {}'.format(self.token),
                   'Content-type': 'application/json'}
        if body:
            body = json.dumps(body)

        connection.request(method, url, body=body, headers=headers)
        response = connection.getresponse()
        if response.status != 200:
            raise HTTPSError('{}: {}'.format(response.status, response.reason ))

        response_data = response.read()
        return json.loads(response_data.decode())

    

    def import_file(self):
        url = '/api/v1/products/import_file'
        method = 'POST'
        with open('123.xlsx', mode='rb') as file: 
            file_content = file.read()
        
        body = {
            'file': file_content,
            'data':
            { 
            'force_update': 'true',
            'only_available': 'false'
             }
        }
            
        

        return self.make_request(method, url, body)

    def get_import_status(self,import_id):
        url = '/api/v1/products/import/status/{}'.format(import_id)
        method = 'GET'

        return self.make_request(method, url)


def main():
    # Initialize Client
    if not AUTH_TOKEN:
        raise Exception('Sorry, there\'s no any AUTH_TOKEN!')

    api_example = EvoClientExample(AUTH_TOKEN)
    
    importfile = api_example.import_file()
    
    print(importfile)
    
    import_id = importurl.get('id')

    import_status = api_example.get_import_status(import_id)

    print(import_status)



if __name__ == '__main__':
    main()

Ошибка:

Traceback (most recent call last):
  File "C:/Users/Programming/Desktop/prom_api 2.py", line 98, in <module>
    main()
  File "C:/Users/Programming/Desktop/prom_api 2.py", line 69, in main
    importfile = api_example.import_file()
  File "C:/Users/Programming/Desktop/prom_api 2.py", line 53, in import_file
    return self.make_request(method, url, body)
  File "C:/Users/Programming/Desktop/prom_api 2.py", line 26, in make_request
    body = json.dumps(body)
  File "C:\Users\Programming\AppData\Local\Programs\Python\Python38\lib\json\__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "C:\Users\Programming\AppData\Local\Programs\Python\Python38\lib\json\encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "C:\Users\Programming\AppData\Local\Programs\Python\Python38\lib\json\encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "C:\Users\Programming\AppData\Local\Programs\Python\Python38\lib\json\encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type bytes is not JSON serializable

Answer 1

Бодался с этим же вопросом. Отмечу своё решение (основные куски кода), что бы люди не страдали. Вся суть в multipart/form-data

url = '/api/v1/products/import_file'
method = 'POST'
params = {
    "force_update": True,
    "only_available": True,
    "mark_missing_product_as": "none",
    "updated_fields": ["price", "presence", "keywords", "description"]
}
files = {"file": "C:\\prod_test.csv"}

def multipart_encoder(params, files):
    boundry = uuid.uuid4().hex
    lines = list()

    lines.append('--' + boundry)
    lines.append('Content-Disposition: form-data; name="data"')
    lines.extend([ '', json.dumps(params) ])

    for key, uri in files.items():
        name = os.path.basename(uri)
        mime = mimetypes.guess_type(uri)[0] or 'application/octet-stream'

        lines.append('--' + boundry)
        lines.append('Content-Disposition: form-data; name="{0}"; filename="{1}"'.format(key, name))
        lines.append('Content-Type: ' + mime)
        lines.append('')
        lines.append(open(uri, 'rb').read())

    lines.append('--%s--'%boundry)

    body = bytes()
    for l in lines:
        if isinstance(l, bytes): body += l + b'\r\n'
        else: body += bytes(l, encoding='utf8') + b'\r\n'

    content_type = 'multipart/form-data; boundary=' + boundry

    return content_type, body

content_type, body = multipart_encoder(params, {"file": file_path})
...
connection = http.client.HTTPSConnection(HOST)
headers = {
    'Authorization': 'Bearer {}'.format(Token),
    'Content-type': content_type,
}
connection.request(method, url, body=body, headers=headers)
response = connection.getresponse()

БЛОГ НА HUSL

string($binary)

Ответы (1 шт):