# -*- coding: utf-8 -*-
import json
import ndjson
from . import utils
[docs]class JsonHandler(FormatHandler):
"""Handle JSON data.
:param str mime_type: the MIME type for the format
:param decoder: the decoder to use for the JSON format
:type decoder: :class:`json.JSONDecoder`
"""
def __init__(self, mime_type, decoder=json.JSONDecoder):
super().__init__(mime_type=mime_type)
self.decoder = decoder
[docs] def parse(self, response):
"""Parse all JSON data from a response.
:param response: raw response
:type response: :class:`requests.Response`
:return: response data
:rtype: JSON
"""
return response.json(cls=self.decoder)
[docs] def parse_stream(self, response):
"""Yield the parsed data from a stream response.
:param response: raw response
:type response: :class:`requests.Response`
:return: iterator over multiple JSON objects
"""
for line in response.iter_lines():
if line:
decoded_line = line.decode('utf-8')
yield json.loads(decoded_line)
[docs]class PgnHandler(FormatHandler):
"""Handle PGN data."""
def __init__(self):
super().__init__(mime_type='application/x-chess-pgn')
[docs] def handle(self, *args, **kwargs):
kwargs['converter'] = utils.noop # disable conversions
return super().handle(*args, **kwargs)
[docs] def parse(self, response):
"""Parse all text data from a response.
:param response: raw response
:type response: :class:`requests.Response`
:return: response text
:rtype: str
"""
return response.text
[docs] def parse_stream(self, response):
"""Yield the parsed PGN games from a stream response.
:param response: raw response
:type response: :class:`requests.Response`
:return: iterator over multiple PGN texts
"""
lines = []
last_line = True
for line in response.iter_lines():
decoded_line = line.decode('utf-8')
if last_line or decoded_line:
lines.append(decoded_line)
else:
yield '\n'.join(lines).strip()
lines = []
last_line = decoded_line
if lines:
yield '\n'.join(lines).strip()
[docs]class TextHandler(FormatHandler):
def __init__(self):
super().__init__(mime_type='text/plain')
[docs] def parse(self, response):
return response.text
[docs] def parse_stream(self, response):
yield from response.iter_lines()
#: Basic text
TEXT = TextHandler()
#: Handles vanilla JSON
JSON = JsonHandler(mime_type='application/json')
#: Handles oddball LiChess JSON (normal JSON, crazy MIME type)
LIJSON = JsonHandler(mime_type='application/vnd.lichess.v3+json')
#: Handles newline-delimited JSON
NDJSON = JsonHandler(mime_type='application/x-ndjson', decoder=ndjson.Decoder)
#: Handles PGN
PGN = PgnHandler()