流式传输时我不断收到 JSONDecodeError

Question

我正在使用 Python 请求库来传输 url 并不断收到此错误：

import json
import requests


s = requests.Session()
payload = {'limit': 0}
r = s.get('https://api.coinmarketcap.com/v1/ticker', params=payload, timeout=(connect_timeout, read_timeout), stream=True)
r.raise_for_status()

for raw_rsvp in r.iter_lines(decode_unicode=True):
    if raw_rsvp:
        rsvp = json.loads(raw_rsvp)
        print(rsvp)

我知道它试图在数组中的对象完成之前读取json，我该如何解决这个问题？

Expecting value: line 1 column 2 (char 1) Traceback (most recent call last):
  File "benchmarks_generator.py", line 26, in refresh_coinmarketcap
    rsvp = json.loads(raw_rsvp)
  File "/usr/local/Cellar/python/3.6.5/Frameworks/Python.framework/Versions/3.6/lib/python3.6/json/__init__.py", line 354, in loads
    return _default_decoder.decode(s)
  File "/usr/local/Cellar/python/3.6.5/Frameworks/Python.framework/Versions/3.6/lib/python3.6/json/decoder.py", line 339, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/local/Cellar/python/3.6.5/Frameworks/Python.framework/Versions/3.6/lib/python3.6/json/decoder.py", line 357, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 2 (char 1)

Answer 1

正如您所说，您收到JSONDecodeError 是因为b'[' 和其他无效json 字符串。并且只要结果是一个 JSON 对象数组，就必须获取整个数组才能将其转换为 python obj。

所以如果你必须使用流式请求（stream=True），唯一的办法就是将输出读取到一个字符串缓冲区（或其他类型的缓冲区来存储临时数据）并在传输时完成，将其转换为JSON obj（这个和其他例子都写在python3）：

import json
import requests

s = requests.Session()
payload = {'limit': 0}
r = s.get('https://api.coinmarketcap.com/v1/ticker', params=payload, stream=True)
r.raise_for_status()

buf = ''
for raw_rsvp in r.iter_lines(decode_unicode=True):
    buf += raw_rsvp.decode()  # bytes -> str for python3 compatibility

rsvp = json.loads(buf)

另一种解决方法是从流中捕获每个 JSON obj 并将其转换为 python obj，然后将它们一一放入 list。但这有点奇怪和慢：

import json
import requests

s = requests.Session()
payload = {'limit': 0}
r = s.get('https://api.coinmarketcap.com/v1/ticker', params=payload, stream=True)
r.raise_for_status()

buf = ''
rsvp = []
for raw_rsvp in r.iter_lines(decode_unicode=True):
    line = raw_rsvp.decode().strip()

    if line in ('[', ']'):
        pass
    elif line == '{':
        buf += line
    elif line in ('}', '},'):
        buf += line
        if buf.endswith(','):
            buf = buf[:-1]  # trim trailing comma
        rsvp.append(json.loads(buf))
        buf = ''  # erase buffer
    else:
        buf += line

P.S.：非流式请求会容易得多：

import requests

s = requests.Session()
payload = {'limit': 0}
r = s.get('https://api.coinmarketcap.com/v1/ticker', params=payload)
r.raise_for_status()
rsvp = r.json()