本文整理汇总了Python中pulsar.utils.system.json.loads函数的典型用法代码示例。如果您正苦于以下问题：Python loads函数的具体用法？Python loads怎么用？Python loads使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了loads函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_rpc
def test_rpc(self):
'''Send a message to the rpc'''
loop = self.http._loop
ws = yield from self.http.get(self.ws, websocket_handler=Message(loop))
self.assertEqual(ws.status_code, 101)
ws.write('Hello there!')
data = yield from ws.handler.get()
data = json.loads(data)
self.assertEqual(data['message'], 'Hello there!')
result = yield from self.rpc.message('Hi!')
self.assertEqual(result, 'OK')
data = yield from ws.handler.get()
data = json.loads(data)
self.assertEqual(data['message'], 'Hi!')
开发者ID:Danzeer，项目名称:pulsar，代码行数:14，代码来源:tests.py
示例2: test_rpc
def test_rpc(self):
'''Send a message to the rpc'''
ws = yield self.http.get(self.ws,
websocket_handler=MessageHandler()
).on_headers
self.assertEqual(ws.handshake.status_code, 101)
ws.write('Hello there!')
data = yield ws.handler.get()
data = json.loads(data)
self.assertEqual(data['message'], 'Hello there!')
result = yield self.rpc.message('Hi!')
self.assertEqual(result, 'OK')
data = yield ws.handler.get()
data = json.loads(data)
self.assertEqual(data['message'], 'Hi!')
开发者ID:BazookaShao，项目名称:pulsar，代码行数:15，代码来源:tests.py
示例3: _ready
def _ready(self, data):
self.environ['wsgi.input'] = BytesIO(data)
try:
self.result = (json.loads(data.decode(self.charset)), None)
except Exception as exc:
raise BadRequest('Could not decode JSON') from exc
return self.result
开发者ID:kakamessi99，项目名称:pulsar-1，代码行数:7，代码来源:formdata.py
示例4: _data_and_files
def _data_and_files(self, data=True, files=True, future=None):
result = {}, None
chunk = None
if future is None:
stream = self.environ.get('wsgi.input')
if self.method not in ENCODE_URL_METHODS and stream:
chunk = stream.read()
if isinstance(chunk, Future):
return chain_future(
chunk, partial(self._data_and_files, data, files))
else:
chunk = future
if chunk is not None:
content_type, options = self.content_type_options
charset = options.get('charset', 'utf-8')
if content_type in JSON_CONTENT_TYPES:
result = json.loads(chunk.decode(charset)), None
elif content_type:
self.environ['wsgi.input'] = BytesIO(chunk)
result = parse_form_data(self.environ, charset)
else:
result = chunk, None
self.environ['wsgi.input'] = BytesIO(chunk)
self.cache.data_and_files = result
return self.data_and_files(data, files)
开发者ID:artemmus，项目名称:pulsar，代码行数:25，代码来源:wrappers.py
示例5: process_data
def process_data(self, response, **kw):
'''Callback passed to :class:`HttpClient` for processing
streaming data.
'''
if response.status_code == 200:
messages = []
data = response.recv_body()
while data:
idx = data.find(b'\r\n')
if idx < 0: # incomplete data - add to buffer
self.buffer.append(data)
data = None
else:
self.buffer.append(data[:idx])
data = data[idx+2:]
msg = b''.join(self.buffer)
self.buffer = []
if msg:
body = json.loads(msg.decode('utf-8'))
if 'disconnect' in body:
msg = body['disconnect']
self.logger.warning('Disconnecting (%d): %s',
msg['code'], msg['reason'])
elif 'warning' in body:
message = body['warning']['message']
self.logger.warning(message)
else:
messages.append(body)
if messages:
# a list of messages is available
if self.cfg.callable:
self.cfg.callable(self, messages)
开发者ID:Danzeer，项目名称:pulsar，代码行数:32，代码来源:twitter.py
示例6: parse_frame
def parse_frame(self, websocket, frame):
parser = frame_parser(kind=1)
frame = parser.decode(frame)
wsclient = websocket.cache.wsclient
websocket.connection.reset_mock()
msg = _json.loads(frame.body[1:])[0]
return wsclient.protocol.decode(msg)
开发者ID:quantmind，项目名称:lux，代码行数:7，代码来源:test.py
示例7: design_delete
def design_delete(self, name):
'''Delete an existing design document at ``name``.
'''
response = yield self.request('head', self._database, '_design', name)
if response.status_code == 200:
rev = json.loads(response.headers['etag'])
yield self.request('delete', self._database, '_design', name,
rev=rev)
开发者ID:huobao36，项目名称:pulsar，代码行数:8，代码来源:store.py
示例8: on_message
def on_message(self, websocket, msg):
request = websocket.request
client = request.cache.mailclient
if msg and client:
msg = json.loads(msg)
if 'mailbox' in msg:
mailbox = yield client.examine(msg['mailbox'])
self.write(request, json.dumps({'mailbox': mailbox}))
开发者ID:JinsongBian，项目名称:pulsar，代码行数:8，代码来源:manage.py
示例9: test_websocket
def test_websocket(self):
ws = yield self.http.get(self.ws, websocket_handler=MessageHandler())
response = ws.handshake
self.assertEqual(response.status_code, 101)
self.assertEqual(response.headers["upgrade"], "websocket")
self.assertEqual(response.connection, ws.connection)
self.assertTrue(ws.connection)
self.assertIsInstance(ws.handler, MessageHandler)
#
data = yield ws.handler.get()
data = json.loads(data)
self.assertEqual(data["message"], "joined")
#
ws.write("Hello there!")
data = yield ws.handler.get()
data = json.loads(data)
self.assertEqual(data["message"], "Hello there!")
开发者ID:huobao36，项目名称:pulsar，代码行数:17，代码来源:app.py
示例10: test_websocket
def test_websocket(self):
ws = yield self.http.get(self.ws, websocket_handler=MessageHandler()
).on_headers
self.assertTrue(ws)
self.assertIsInstance(ws.handler, MessageHandler)
ws.write('Hello there!')
data = yield ws.handler.get()
data = json.loads(data)
self.assertEqual(data['message'], 'Hello there!')
开发者ID:BazookaShao，项目名称:pulsar，代码行数:9，代码来源:app.py
示例11: test_websocket
def test_websocket(self):
c = self.http
ws = yield c.get(self.ws, websocket_handler=MessageHandler(c._loop))
response = ws.handshake
self.assertEqual(response.status_code, 101)
self.assertEqual(response.headers['upgrade'], 'websocket')
self.assertEqual(response.connection, ws.connection)
self.assertTrue(ws.connection)
self.assertIsInstance(ws.handler, MessageHandler)
#
data = yield ws.handler.get()
data = json.loads(data)
self.assertEqual(data['message'], 'joined')
#
ws.write('Hello there!')
data = yield ws.handler.get()
data = json.loads(data)
self.assertEqual(data['message'], 'Hello there!')
开发者ID:axisofentropy，项目名称:pulsar，代码行数:18，代码来源:app.py
示例12: to_python
def to_python(self, value, store=None):
if value is None:
return self.get_default()
if isinstance(value, str):
try:
return json.loads(value)
except TypeError:
return None
else:
return value
开发者ID:axisofentropy，项目名称:pulsar，代码行数:10，代码来源:fields.py
示例13: _data_and_files
def _data_and_files(self):
result = {}, None
stream = self.environ.get('wsgi.input')
try:
if self.method not in ENCODE_URL_METHODS and stream:
chunk = yield stream.read()
content_type, options = self.content_type_options
charset = options.get('charset', 'utf-8')
if content_type in JSON_CONTENT_TYPES:
result = json.loads(chunk.decode(charset)), None
else:
self.environ['wsgi.input'] = BytesIO(chunk)
result = parse_form_data(self.environ, charset)
# set the wsgi.input to a readable file-like object for
# third-parties application (django or any other web-framework)
self.environ['wsgi.input'] = BytesIO(chunk)
finally:
self._cached_data_and_files = result
coroutine_return(result)
开发者ID:BazookaShao，项目名称:pulsar，代码行数:19，代码来源:wrappers.py
示例14: media_libraries
def media_libraries():
global _media_libraries
if _media_libraries is None:
if os.path.isfile('libs.json'): # pragma nocover
with open('libs.json') as f:
data = f.read()
_media_libraries = json.loads(data)
else:
from pulsar import new_event_loop
from pulsar.apps.http import HttpClient
http = HttpClient(loop=new_event_loop())
try:
response = http.get(_libs_url)
_media_libraries = response.json()
except Exception: # pragma nocover
http.logger.error('Could not import media library',
exc_info=True)
_media_libraries = {'libs': {}, 'deps': {}}
return _media_libraries
开发者ID:axisofentropy，项目名称:pulsar，代码行数:19，代码来源:content.py
示例15: _data_and_files
def _data_and_files(self, data=True, files=True):
result = {}, None
stream = self.environ.get("wsgi.input")
chunk = None
try:
if self.method not in ENCODE_URL_METHODS and stream:
chunk = stream.read()
if isinstance(chunk, Future):
chunk = yield chunk
content_type, options = self.content_type_options
charset = options.get("charset", "utf-8")
if content_type in JSON_CONTENT_TYPES:
result = json.loads(chunk.decode(charset)), None
else:
self.environ["wsgi.input"] = BytesIO(chunk)
result = parse_form_data(self.environ, charset)
finally:
self.cache.data_and_files = result
if chunk is not None:
self.environ["wsgi.input"] = BytesIO(chunk)
coroutine_return(self.data_and_files(data, files))
开发者ID:Ghost-script，项目名称:dyno-chat，代码行数:21，代码来源:wrappers.py
示例16: json
def json(self, charset=None):
'''Decode content as a JSON object.'''
return json.loads(self.content_string(charset))
开发者ID:successtest9，项目名称:pulsar，代码行数:3，代码来源:__init__.py
示例17: json
def json(self, charset=None):
"""Decode content as a JSON object.
"""
return json.loads(self.text(charset))
开发者ID:arhik，项目名称:pulsar，代码行数:4，代码来源:__init__.py
示例18: _ready
def _ready(self, data):
self.environ["wsgi.input"] = BytesIO(data)
self.result = (json.loads(data.decode(self.charset)), None)
return self.result
开发者ID:Pastafarianist，项目名称:pulsar，代码行数:4，代码来源:formdata.py
示例19: load
def load(cls, data, method=None):
method = method or 'json'
if method == 'json':
return cls(**json.loads(to_string(data)))
else:
raise ImproperlyConfigured('Unknown serialisation "%s"' % method)
开发者ID:dejlek，项目名称:pulsar-queue，代码行数:6，代码来源:task.py
示例20: _ready
def _ready(self, data):
self.result = (json.loads(data.decode(self.charset)), None)
return self.result
开发者ID:successtest9，项目名称:pulsar，代码行数:3，代码来源:formdata.py
注：本文中的pulsar.utils.system.json.loads函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论