def split_buffer(stream, splitter=None, decoder=lambda a: a):
"""Given a generator which yields strings and a splitter function,
joins all input, splits on the separator and yields each chunk.
Unlike string.split(), each chunk includes the trailing
separator, except for the last one if none was found on the end
of the input.
"""
splitter = splitter or line_splitter
buffered = six.text_type('')
for data in stream_as_text(stream):
buffered += data
while True:
buffer_split = splitter(buffered)
if buffer_split is None:
break
item, buffered = buffer_split
yield item
if buffered:
yield decoder(buffered)
python类decoder()的实例源码
def embed_from_json(self, ctx, *, js_text: str):
"""Send an embed from JSON.
Usage: embed_from_json [json]"""
echeck_perms(ctx, ('bot_owner',))
class SemiEmbed:
def __init__(self, obj):
self.obj = obj
def to_dict(self):
return self.obj
try:
embed_obj = json.loads(js_text)
except json.decoder.JSONDecodeError:
await ctx.send(':warning: **Invalid JSON data!**')
else:
sembed = SemiEmbed(embed_obj)
try:
await ctx.send(embed=sembed)
except discord.HTTPException as e:
if '400' in str(e):
await ctx.send(':warning: **Couldn\'t send embed, check your data!**')
else:
raise e
def split_buffer(stream, splitter=None, decoder=lambda a: a):
"""Given a generator which yields strings and a splitter function,
joins all input, splits on the separator and yields each chunk.
Unlike string.split(), each chunk includes the trailing
separator, except for the last one if none was found on the end
of the input.
"""
splitter = splitter or line_splitter
buffered = six.text_type('')
for data in stream_as_text(stream):
buffered += data
while True:
buffer_split = splitter(buffered)
if buffer_split is None:
break
item, buffered = buffer_split
yield item
if buffered:
try:
yield decoder(buffered)
except Exception as e:
raise StreamParseError(e)
def json_splitter(buffer):
"""Attempt to parse a json object from a buffer. If there is at least one
object, return it and the rest of the buffer, otherwise return None.
"""
try:
obj, index = json_decoder.raw_decode(buffer)
rest = buffer[json.decoder.WHITESPACE.match(buffer, index).end():]
return obj, rest
except ValueError:
return None
def test_pyjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__,
'json.scanner')
self.assertEqual(self.json.decoder.scanstring.__module__,
'json.decoder')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'json.encoder')
def test_cjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__, '_json')
self.assertEqual(self.json.decoder.scanstring.__module__, '_json')
self.assertEqual(self.json.encoder.c_make_encoder.__module__, '_json')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'_json')
def additional_tests():
suite = unittest.TestSuite()
for mod in (json, json.encoder, json.decoder):
suite.addTest(doctest.DocTestSuite(mod))
suite.addTest(TestPyTest('test_pyjson'))
suite.addTest(TestCTest('test_cjson'))
return suite
def loads(self, str_data):
try:
decoder = import_string(settings.MODERNRPC_JSON_DECODER)
return json.loads(str_data, cls=decoder)
except JSONDecodeError as e:
raise RPCParseError(str(e))
def test_pyjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__,
'json.scanner')
self.assertEqual(self.json.decoder.scanstring.__module__,
'json.decoder')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'json.encoder')
def test_cjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__, '_json')
self.assertEqual(self.json.decoder.scanstring.__module__, '_json')
self.assertEqual(self.json.encoder.c_make_encoder.__module__, '_json')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'_json')
def additional_tests():
suite = unittest.TestSuite()
for mod in (json, json.encoder, json.decoder):
suite.addTest(doctest.DocTestSuite(mod))
suite.addTest(TestPyTest('test_pyjson'))
suite.addTest(TestCTest('test_cjson'))
return suite
def test_pyjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__,
'json.scanner')
self.assertEqual(self.json.decoder.scanstring.__module__,
'json.decoder')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'json.encoder')
def test_cjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__, '_json')
self.assertEqual(self.json.decoder.scanstring.__module__, '_json')
self.assertEqual(self.json.encoder.c_make_encoder.__module__, '_json')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'_json')
def additional_tests():
suite = unittest.TestSuite()
for mod in (json, json.encoder, json.decoder):
suite.addTest(doctest.DocTestSuite(mod))
suite.addTest(TestPyTest('test_pyjson'))
suite.addTest(TestCTest('test_cjson'))
return suite
def test_pyjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__,
'json.scanner')
self.assertEqual(self.json.decoder.scanstring.__module__,
'json.decoder')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'json.encoder')
def test_cjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__, '_json')
self.assertEqual(self.json.decoder.scanstring.__module__, '_json')
self.assertEqual(self.json.encoder.c_make_encoder.__module__, '_json')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'_json')
def additional_tests():
suite = unittest.TestSuite()
for mod in (json, json.encoder, json.decoder):
suite.addTest(doctest.DocTestSuite(mod))
suite.addTest(TestPyTest('test_pyjson'))
suite.addTest(TestCTest('test_cjson'))
return suite
def json_splitter(buffer):
"""Attempt to parse a json object from a buffer. If there is at least one
object, return it and the rest of the buffer, otherwise return None.
"""
buffer = buffer.strip()
try:
obj, index = json_decoder.raw_decode(buffer)
rest = buffer[json.decoder.WHITESPACE.match(buffer, index).end():]
return obj, rest
except ValueError:
return None
def log_in_for_userkey():
headers = {}
url = settings.API_LOGIN_URL + settings.APPKEY
provided_api_app = settings.SECRETKEY and settings.ACCOUNTKEY
while True:
username = get_username()
if provided_api_app:
data = {'login': username, 'accountkey': settings.ACCOUNTKEY}
headers = apisign(url, settings.SECRETKEY, **data)
msg = 'Nieprawid?owy login, accountkey, lub secretkey.'
else:
password = get_password()
data = {'login': username, 'password': password}
msg = 'Nieprawid?owy login lub has?o.'
response = requests.post(url, data=data, headers=headers)
password = None
data = None
try:
userkey = response.json()['userkey']
except (json.decoder.JSONDecodeError, KeyError):
logging.error(msg)
logging.debug(traceback.format_exc())
if provided_api_app:
raise SystemExit
reset_credentials()
continue
else:
settings.USERKEY = userkey
save_userkey(userkey)
break
def test_pyjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__,
'json.scanner')
self.assertEqual(self.json.decoder.scanstring.__module__,
'json.decoder')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'json.encoder')
def test_cjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__, '_json')
self.assertEqual(self.json.decoder.scanstring.__module__, '_json')
self.assertEqual(self.json.encoder.c_make_encoder.__module__, '_json')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'_json')
def additional_tests():
suite = unittest.TestSuite()
for mod in (json, json.encoder, json.decoder):
suite.addTest(doctest.DocTestSuite(mod))
suite.addTest(TestPyTest('test_pyjson'))
suite.addTest(TestCTest('test_cjson'))
return suite
def test_pyjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__,
'json.scanner')
self.assertEqual(self.json.decoder.scanstring.__module__,
'json.decoder')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'json.encoder')
def test_cjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__, '_json')
self.assertEqual(self.json.decoder.scanstring.__module__, '_json')
self.assertEqual(self.json.encoder.c_make_encoder.__module__, '_json')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'_json')
def load_tests(loader, _, pattern):
suite = unittest.TestSuite()
for mod in (json, json.encoder, json.decoder):
suite.addTest(doctest.DocTestSuite(mod))
suite.addTest(TestPyTest('test_pyjson'))
suite.addTest(TestCTest('test_cjson'))
pkg_dir = os.path.dirname(__file__)
return support.load_package_tests(pkg_dir, loader, suite, pattern)
def test_pyjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__,
'json.scanner')
self.assertEqual(self.json.decoder.scanstring.__module__,
'json.decoder')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'json.encoder')
def test_cjson(self):
self.assertEqual(self.json.scanner.make_scanner.__module__, '_json')
self.assertEqual(self.json.decoder.scanstring.__module__, '_json')
self.assertEqual(self.json.encoder.c_make_encoder.__module__, '_json')
self.assertEqual(self.json.encoder.encode_basestring_ascii.__module__,
'_json')
def additional_tests():
suite = unittest.TestSuite()
for mod in (json, json.encoder, json.decoder):
suite.addTest(doctest.DocTestSuite(mod))
suite.addTest(TestPyTest('test_pyjson'))
suite.addTest(TestCTest('test_cjson'))
return suite
def decoder(jsonobj):
if 'source' in jsonobj:
return TrackedLayer(jsonobj['source'],
jsonobj['repoUrl'])
else:
return jsonobj
def readTrackedLayers():
try:
global tracked
filename = os.path.join(userFolder(), "trackedlayers")
if os.path.exists(filename):
with open(filename) as f:
lines = f.readlines()
jsonstring = "\n".join(lines)
if jsonstring:
tracked = JSONDecoder(object_hook = decoder).decode(jsonstring)
except KeyError:
pass
