KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use quoted_string_type method in Radish
Best Python code snippet using radish
typed_inserter.py
Source:typed_inserter.py 
1# encoding: utf-82#3#4# This Source Code Form is subject to the terms of the Mozilla Public5# License, v. 2.0. If a copy of the MPL was not distributed with this file,6# You can obtain one at http://mozilla.org/MPL/2.0/.7#8# Author: Kyle Lahnakoski (kyle@lahnakoski.com)9#10from __future__ import absolute_import11from __future__ import division12from __future__ import unicode_literals13import json14import time15from collections import Mapping16from datetime import datetime, date, timedelta17from decimal import Decimal18from jx_base import python_type_to_json_type, INTEGER, NUMBER, EXISTS, NESTED, STRING, BOOLEAN, OBJECT19from jx_python.expressions import jx_expression_to_function20from jx_python.meta import Column21from mo_dots import Data, FlatList, NullType, unwrap22from mo_future import text_type, binary_type, utf8_json_encoder, long, sort_using_key23from mo_json import ESCAPE_DCT, float2json, json2value24from mo_json.encoder import problem_serializing, UnicodeBuilder, COMMA, COLON25from mo_json.typed_encoder import encode_property, BOOLEAN_TYPE, NESTED_TYPE, EXISTS_TYPE, STRING_TYPE, NUMBER_TYPE26from mo_logs import Log27from mo_logs.strings import utf82unicode, quote28from mo_times.dates import Date29from mo_times.durations import Duration30from pyLibrary.env.elasticsearch import parse_properties, random_id, es_type_to_json_type31append = UnicodeBuilder.append32QUOTED_BOOLEAN_TYPE = quote(BOOLEAN_TYPE)33QUOTED_NUMBER_TYPE = quote(NUMBER_TYPE)34QUOTED_STRING_TYPE = quote(STRING_TYPE)35QUOTED_NESTED_TYPE = quote(NESTED_TYPE)36QUOTED_EXISTS_TYPE = quote(EXISTS_TYPE)37json_type_to_inserter_type = {38 BOOLEAN: BOOLEAN_TYPE,39 INTEGER: NUMBER_TYPE,40 NUMBER: NUMBER_TYPE,41 STRING: STRING_TYPE,42 NESTED: NESTED_TYPE,43 EXISTS: EXISTS_TYPE44}45class TypedInserter(object):46 def __init__(self, es=None, id_expression="_id"):47 self.es = es48 self.id_column = id_expression49 self.get_id = jx_expression_to_function(id_expression)50 self.remove_id = True if id_expression == "_id" else False51 if es:52 _schema = Data()53 for c in parse_properties(es.settings.alias, ".", es.get_properties()):54 if c.es_type not in (OBJECT, NESTED):55 _schema[c.names["."]] = c56 self.schema = unwrap(_schema)57 else:58 self.schema = {}59 def typed_encode(self, r):60 """61 :param record: expecting id and value properties62 :return: dict with id and json properties63 """64 try:65 value = r['value']66 if "json" in r:67 value = json2value(r["json"])68 elif isinstance(value, Mapping) or value != None:69 pass70 else:71 from mo_logs import Log72 raise Log.error("Expecting every record given to have \"value\" or \"json\" property")73 _buffer = UnicodeBuilder(1024)74 net_new_properties = []75 path = []76 if isinstance(value, Mapping):77 given_id = self.get_id(value)78 if self.remove_id:79 value['_id'] = None80 else:81 given_id = None82 if given_id:83 record_id = r.get('id')84 if record_id and record_id != given_id:85 from mo_logs import Log86 raise Log.error(87 "expecting {{property}} of record ({{record_id|quote}}) to match one given ({{given|quote}})",88 property=self.id_column,89 record_id=record_id,90 given=given_id91 )92 else:93 record_id = r.get('id')94 if record_id:95 given_id = record_id96 else:97 given_id = random_id()98 self._typed_encode(value, self.schema, path, net_new_properties, _buffer)99 json = _buffer.build()100 for props in net_new_properties:101 path, type = props[:-1], props[-1][1:]102 # self.es.add_column(join_field(path), type)103 return {"id": given_id, "json": json}104 except Exception as e:105 # THE PRETTY JSON WILL PROVIDE MORE DETAIL ABOUT THE SERIALIZATION CONCERNS106 from mo_logs import Log107 Log.error("Serialization of JSON problems", cause=e)108 def _typed_encode(self, value, sub_schema, path, net_new_properties, _buffer):109 try:110 if isinstance(sub_schema, Column):111 value_json_type = python_type_to_json_type[value.__class__]112 column_json_type = es_type_to_json_type[sub_schema.es_type]113 if value_json_type == column_json_type:114 pass # ok115 elif value_json_type == NESTED and all(python_type_to_json_type[v.__class__] == column_json_type for v in value if v != None):116 pass # empty arrays can be anything117 else:118 from mo_logs import Log119 Log.error("Can not store {{value}} in {{column|quote}}", value=value, column=sub_schema.names['.'])120 sub_schema = {json_type_to_inserter_type[value_json_type]: sub_schema}121 if value is None:122 append(_buffer, '{}')123 return124 elif value is True:125 if BOOLEAN_TYPE not in sub_schema:126 sub_schema[BOOLEAN_TYPE] = {}127 net_new_properties.append(path+[BOOLEAN_TYPE])128 append(_buffer, '{'+QUOTED_BOOLEAN_TYPE+COLON+'true}')129 return130 elif value is False:131 if BOOLEAN_TYPE not in sub_schema:132 sub_schema[BOOLEAN_TYPE] = {}133 net_new_properties.append(path+[BOOLEAN_TYPE])134 append(_buffer, '{'+QUOTED_BOOLEAN_TYPE+COLON+'false}')135 return136 _type = value.__class__137 if _type in (dict, Data):138 if isinstance(sub_schema, Column):139 from mo_logs import Log140 Log.error("Can not handle {{column|json}}", column=sub_schema)141 if NESTED_TYPE in sub_schema:142 # PREFER NESTED, WHEN SEEN BEFORE143 if value:144 append(_buffer, '{'+QUOTED_NESTED_TYPE+COLON+'[')145 self._dict2json(value, sub_schema[NESTED_TYPE], path + [NESTED_TYPE], net_new_properties, _buffer)146 append(_buffer, ']'+COMMA+QUOTED_EXISTS_TYPE+COLON + text_type(len(value)) + '}')147 else:148 # SINGLETON LISTS OF null SHOULD NOT EXIST149 from mo_logs import Log150 Log.error("should not happen")151 else:152 if EXISTS_TYPE not in sub_schema:153 sub_schema[EXISTS_TYPE] = {}154 net_new_properties.append(path+[EXISTS_TYPE])155 if value:156 self._dict2json(value, sub_schema, path, net_new_properties, _buffer)157 else:158 append(_buffer, '{'+QUOTED_EXISTS_TYPE+COLON+'0}')159 elif _type is binary_type:160 if STRING_TYPE not in sub_schema:161 sub_schema[STRING_TYPE] = True162 net_new_properties.append(path + [STRING_TYPE])163 append(_buffer, '{'+QUOTED_STRING_TYPE+COLON+'"')164 try:165 v = utf82unicode(value)166 except Exception as e:167 raise problem_serializing(value, e)168 for c in v:169 append(_buffer, ESCAPE_DCT.get(c, c))170 append(_buffer, '"}')171 elif _type is text_type:172 if STRING_TYPE not in sub_schema:173 sub_schema[STRING_TYPE] = True174 net_new_properties.append(path + [STRING_TYPE])175 append(_buffer, '{'+QUOTED_STRING_TYPE+COLON+'"')176 for c in value:177 append(_buffer, ESCAPE_DCT.get(c, c))178 append(_buffer, '"}')179 elif _type in (int, long, Decimal):180 if NUMBER_TYPE not in sub_schema:181 sub_schema[NUMBER_TYPE] = True182 net_new_properties.append(path + [NUMBER_TYPE])183 append(_buffer, '{'+QUOTED_NUMBER_TYPE+COLON)184 append(_buffer, float2json(value))185 append(_buffer, '}')186 elif _type is float:187 if NUMBER_TYPE not in sub_schema:188 sub_schema[NUMBER_TYPE] = True189 net_new_properties.append(path + [NUMBER_TYPE])190 append(_buffer, '{'+QUOTED_NUMBER_TYPE+COLON)191 append(_buffer, float2json(value))192 append(_buffer, '}')193 elif _type in (set, list, tuple, FlatList):194 if len(value) == 0:195 append(_buffer, '{'+QUOTED_NESTED_TYPE+COLON+'[]}')196 elif any(isinstance(v, (Mapping, set, list, tuple, FlatList)) for v in value):197 if NESTED_TYPE not in sub_schema:198 sub_schema[NESTED_TYPE] = {}199 net_new_properties.append(path + [NESTED_TYPE])200 append(_buffer, '{'+QUOTED_NESTED_TYPE+COLON)201 self._list2json(value, sub_schema[NESTED_TYPE], path+[NESTED_TYPE], net_new_properties, _buffer)202 append(_buffer, '}')203 else:204 # ALLOW PRIMITIVE MULTIVALUES205 value = [v for v in value if v != None]206 types = list(set(python_type_to_json_type[v.__class__] for v in value))207 if len(types) == 0: # HANDLE LISTS WITH Nones IN THEM208 append(_buffer, '{'+QUOTED_NESTED_TYPE+COLON+'[]}')209 elif len(types) > 1:210 from mo_logs import Log211 Log.error("Can not handle multi-typed multivalues")212 else:213 element_type = json_type_to_inserter_type[types[0]]214 if element_type not in sub_schema:215 sub_schema[element_type] = True216 net_new_properties.append(path + [element_type])217 append(_buffer, '{'+quote(element_type)+COLON)218 self._multivalue2json(value, sub_schema[element_type], path + [element_type], net_new_properties, _buffer)219 append(_buffer, '}')220 elif _type is date:221 if NUMBER_TYPE not in sub_schema:222 sub_schema[NUMBER_TYPE] = True223 net_new_properties.append(path + [NUMBER_TYPE])224 append(_buffer, '{'+QUOTED_NUMBER_TYPE+COLON)225 append(_buffer, float2json(time.mktime(value.timetuple())))226 append(_buffer, '}')227 elif _type is datetime:228 if NUMBER_TYPE not in sub_schema:229 sub_schema[NUMBER_TYPE] = True230 net_new_properties.append(path + [NUMBER_TYPE])231 append(_buffer, '{'+QUOTED_NUMBER_TYPE+COLON)232 append(_buffer, float2json(time.mktime(value.timetuple())))233 append(_buffer, '}')234 elif _type is Date:235 if NUMBER_TYPE not in sub_schema:236 sub_schema[NUMBER_TYPE] = True237 net_new_properties.append(path + [NUMBER_TYPE])238 append(_buffer, '{'+QUOTED_NUMBER_TYPE+COLON)239 append(_buffer, float2json(value.unix))240 append(_buffer, '}')241 elif _type is timedelta:242 if NUMBER_TYPE not in sub_schema:243 sub_schema[NUMBER_TYPE] = True244 net_new_properties.append(path + [NUMBER_TYPE])245 append(_buffer, '{'+QUOTED_NUMBER_TYPE+COLON)246 append(_buffer, float2json(value.total_seconds()))247 append(_buffer, '}')248 elif _type is Duration:249 if NUMBER_TYPE not in sub_schema:250 sub_schema[NUMBER_TYPE] = True251 net_new_properties.append(path + [NUMBER_TYPE])252 append(_buffer, '{'+QUOTED_NUMBER_TYPE+COLON)253 append(_buffer, float2json(value.seconds))254 append(_buffer, '}')255 elif _type is NullType:256 append(_buffer, 'null')257 elif hasattr(value, '__iter__'):258 if NESTED_TYPE not in sub_schema:259 sub_schema[NESTED_TYPE] = {}260 net_new_properties.append(path + [NESTED_TYPE])261 append(_buffer, '{'+QUOTED_NESTED_TYPE+COLON)262 self._iter2json(value, sub_schema[NESTED_TYPE], path+[NESTED_TYPE], net_new_properties, _buffer)263 append(_buffer, '}')264 else:265 from mo_logs import Log266 Log.error(text_type(repr(value)) + " is not JSON serializable")267 except Exception as e:268 from mo_logs import Log269 Log.error(text_type(repr(value)) + " is not JSON serializable", cause=e)270 def _list2json(self, value, sub_schema, path, net_new_properties, _buffer):271 if not value:272 append(_buffer, '[]')273 else:274 sep = '['275 for v in value:276 append(_buffer, sep)277 sep = COMMA278 self._typed_encode(v, sub_schema, path, net_new_properties, _buffer)279 append(_buffer, ']'+COMMA+QUOTED_EXISTS_TYPE+COLON+text_type(len(value)))280 def _multivalue2json(self, value, sub_schema, path, net_new_properties, _buffer):281 if not value:282 append(_buffer, '[]')283 elif len(value) == 1:284 append(_buffer, json_encoder(value[0]))285 else:286 sep = '['287 for v in value:288 append(_buffer, sep)289 sep = COMMA290 append(_buffer, json_encoder(v))291 append(_buffer, ']')292 def _iter2json(self, value, sub_schema, path, net_new_properties, _buffer):293 append(_buffer, '[')294 sep = ''295 count = 0296 for v in value:297 append(_buffer, sep)298 sep = COMMA299 self._typed_encode(v, sub_schema, path, net_new_properties, _buffer)300 count += 1301 append(_buffer, ']' + COMMA + QUOTED_EXISTS_TYPE + COLON + text_type(count))302 def _dict2json(self, value, sub_schema, path, net_new_properties, _buffer):303 prefix = '{'304 for k, v in sort_using_key(value.items(), lambda r: r[0]):305 if v == None or v == '':306 continue307 append(_buffer, prefix)308 prefix = COMMA309 if isinstance(k, binary_type):310 k = utf82unicode(k)311 if not isinstance(k, text_type):312 Log.error("Expecting property name to be a string")313 if k not in sub_schema:314 sub_schema[k] = {}315 net_new_properties.append(path+[k])316 append(_buffer, json.dumps(encode_property(k)))317 append(_buffer, COLON)318 self._typed_encode(v, sub_schema[k], path+[k], net_new_properties, _buffer)319 if prefix == COMMA:320 append(_buffer, COMMA+QUOTED_EXISTS_TYPE+COLON+'1}')321 else:322 append(_buffer, '{'+QUOTED_EXISTS_TYPE+COLON+'0}')...
typed_encoder.py
Source:typed_encoder.py
1# encoding: utf-82#3#4# This Source Code Form is subject to the terms of the Mozilla Public5# License, v. 2.0. If a copy of the MPL was not distributed with this file,6# You can obtain one at http://mozilla.org/MPL/2.0/.7#8# Author: Kyle Lahnakoski (kyle@lahnakoski.com)9#10from __future__ import absolute_import11from __future__ import division12from __future__ import unicode_literals13from collections import Mapping14from decimal import Decimal15from json.encoder import encode_basestring16from datetime import date, datetime, timedelta17import time18from jx_base import Column, python_type_to_json_type, NESTED, EXISTS, STRING, NUMBER, INTEGER, BOOLEAN19from mo_dots import Data, FlatList, NullType, join_field, split_field20from mo_future import text_type, binary_type, sort_using_key21from mo_logs import Log22from mo_logs.strings import quote, utf82unicode23from mo_times import Date, Duration24from mo_json import ESCAPE_DCT, float2json25from mo_json.encoder import UnicodeBuilder, COLON, COMMA, problem_serializing, json_encoder26def encode_property(name):27 return name.replace(",", "\\,").replace(".", ",")28def decode_property(encoded):29 return encoded.replace("\\,", "\a").replace(",", ".").replace("\a", ",")30def untype_path(encoded):31 if encoded.startswith(".."):32 remainder = encoded.lstrip(".")33 back = len(encoded) - len(remainder) - 134 return ("." * back) + join_field(decode_property(c) for c in split_field(remainder) if not c.startswith(TYPE_PREFIX))35 else:36 return join_field(decode_property(c) for c in split_field(encoded) if not c.startswith(TYPE_PREFIX))37def unnest_path(encoded):38 if encoded.startswith(".."):39 encoded = encoded.lstrip(".")40 if not encoded:41 encoded = "."42 return join_field(decode_property(c) for c in split_field(encoded) if c != NESTED_TYPE)43def untyped(value):44 return _untype(value)45def _untype(value):46 if isinstance(value, Mapping):47 output = {}48 for k, v in value.items():49 if k == EXISTS_TYPE:50 continue51 elif k.startswith(TYPE_PREFIX):52 return v53 else:54 output[decode_property(k)] = _untype(v)55 return output56 elif isinstance(value, list):57 return [_untype(v) for v in value]58 else:59 return value60def encode(value):61 buffer = UnicodeBuilder(1024)62 typed_encode(63 value,64 sub_schema={},65 path=[],66 net_new_properties=[],67 buffer=buffer68 )69 return buffer.build()70def typed_encode(value, sub_schema, path, net_new_properties, buffer):71 """72 :param value: THE DATASCRUTURE TO ENCODE73 :param sub_schema: dict FROM PATH TO Column DESCRIBING THE TYPE74 :param path: list OF CURRENT PATH75 :param net_new_properties: list FOR ADDING NEW PROPERTIES NOT FOUND IN sub_schema76 :param buffer: UnicodeBuilder OBJECT77 :return:78 """79 try:80 if isinstance(sub_schema, Column):81 value_json_type = python_type_to_json_type[value.__class__]82 column_json_type = es_type_to_json_type[sub_schema.es_type]83 if value_json_type == column_json_type:84 pass # ok85 elif value_json_type == NESTED and all(python_type_to_json_type[v.__class__] == column_json_type for v in value if v != None):86 pass # empty arrays can be anything87 else:88 from mo_logs import Log89 Log.error("Can not store {{value}} in {{column|quote}}", value=value, column=sub_schema.names['.'])90 sub_schema = {json_type_to_inserter_type[value_json_type]: sub_schema}91 if value == None:92 from mo_logs import Log93 Log.error("can not encode null (missing) values")94 elif value is True:95 if BOOLEAN_TYPE not in sub_schema:96 sub_schema[BOOLEAN_TYPE] = {}97 net_new_properties.append(path + [BOOLEAN_TYPE])98 append(buffer, '{')99 append(buffer, QUOTED_BOOLEAN_TYPE)100 append(buffer, 'true}')101 return102 elif value is False:103 if BOOLEAN_TYPE not in sub_schema:104 sub_schema[BOOLEAN_TYPE] = {}105 net_new_properties.append(path + [BOOLEAN_TYPE])106 append(buffer, '{')107 append(buffer, QUOTED_BOOLEAN_TYPE)108 append(buffer, 'false}')109 return110 _type = value.__class__111 if _type in (dict, Data):112 if isinstance(sub_schema, Column):113 from mo_logs import Log114 Log.error("Can not handle {{column|json}}", column=sub_schema)115 if NESTED_TYPE in sub_schema:116 # PREFER NESTED, WHEN SEEN BEFORE117 if value:118 append(buffer, '{')119 append(buffer, QUOTED_NESTED_TYPE)120 append(buffer, '[')121 _dict2json(value, sub_schema[NESTED_TYPE], path + [NESTED_TYPE], net_new_properties, buffer)122 append(buffer, ']' + COMMA)123 append(buffer, QUOTED_EXISTS_TYPE)124 append(buffer, text_type(len(value)))125 append(buffer, '}')126 else:127 # SINGLETON LISTS OF null SHOULD NOT EXIST128 from mo_logs import Log129 Log.error("should not happen")130 else:131 if EXISTS_TYPE not in sub_schema:132 sub_schema[EXISTS_TYPE] = {}133 net_new_properties.append(path + [EXISTS_TYPE])134 if value:135 _dict2json(value, sub_schema, path, net_new_properties, buffer)136 else:137 append(buffer, '{')138 append(buffer, QUOTED_EXISTS_TYPE)139 append(buffer, '0}')140 elif _type is binary_type:141 if STRING_TYPE not in sub_schema:142 sub_schema[STRING_TYPE] = True143 net_new_properties.append(path + [STRING_TYPE])144 append(buffer, '{')145 append(buffer, QUOTED_STRING_TYPE)146 append(buffer, '"')147 try:148 v = utf82unicode(value)149 except Exception as e:150 raise problem_serializing(value, e)151 for c in v:152 append(buffer, ESCAPE_DCT.get(c, c))153 append(buffer, '"}')154 elif _type is text_type:155 if STRING_TYPE not in sub_schema:156 sub_schema[STRING_TYPE] = True157 net_new_properties.append(path + [STRING_TYPE])158 append(buffer, '{')159 append(buffer, QUOTED_STRING_TYPE)160 append(buffer, '"')161 for c in value:162 append(buffer, ESCAPE_DCT.get(c, c))163 append(buffer, '"}')164 elif _type in (int, long, Decimal):165 if NUMBER_TYPE not in sub_schema:166 sub_schema[NUMBER_TYPE] = True167 net_new_properties.append(path + [NUMBER_TYPE])168 append(buffer, '{')169 append(buffer, QUOTED_NUMBER_TYPE)170 append(buffer, float2json(value))171 append(buffer, '}')172 elif _type is float:173 if NUMBER_TYPE not in sub_schema:174 sub_schema[NUMBER_TYPE] = True175 net_new_properties.append(path + [NUMBER_TYPE])176 append(buffer, '{')177 append(buffer, QUOTED_NUMBER_TYPE)178 append(buffer, float2json(value))179 append(buffer, '}')180 elif _type in (set, list, tuple, FlatList):181 if len(value) == 0:182 append(buffer, '{')183 append(buffer, QUOTED_NESTED_TYPE)184 append(buffer, '[]}')185 elif any(isinstance(v, (Mapping, set, list, tuple, FlatList)) for v in value):186 if NESTED_TYPE not in sub_schema:187 sub_schema[NESTED_TYPE] = {}188 net_new_properties.append(path + [NESTED_TYPE])189 append(buffer, '{')190 append(buffer, QUOTED_NESTED_TYPE)191 _list2json(value, sub_schema[NESTED_TYPE], path + [NESTED_TYPE], net_new_properties, buffer)192 append(buffer, '}')193 else:194 # ALLOW PRIMITIVE MULTIVALUES195 value = [v for v in value if v != None]196 types = list(set(json_type_to_inserter_type[python_type_to_json_type[v.__class__]] for v in value))197 if len(types) == 0: # HANDLE LISTS WITH Nones IN THEM198 append(buffer, '{')199 append(buffer, QUOTED_NESTED_TYPE)200 append(buffer, '[]}')201 elif len(types) > 1:202 _list2json(value, sub_schema, path + [NESTED_TYPE], net_new_properties, buffer)203 else:204 element_type = types[0]205 if element_type not in sub_schema:206 sub_schema[element_type] = True207 net_new_properties.append(path + [element_type])208 append(buffer, '{')209 append(buffer, quote(element_type))210 append(buffer, COLON)211 _multivalue2json(value, sub_schema[element_type], path + [element_type], net_new_properties, buffer)212 append(buffer, '}')213 elif _type is date:214 if NUMBER_TYPE not in sub_schema:215 sub_schema[NUMBER_TYPE] = True216 net_new_properties.append(path + [NUMBER_TYPE])217 append(buffer, '{')218 append(buffer, QUOTED_NUMBER_TYPE)219 append(buffer, float2json(time.mktime(value.timetuple())))220 append(buffer, '}')221 elif _type is datetime:222 if NUMBER_TYPE not in sub_schema:223 sub_schema[NUMBER_TYPE] = True224 net_new_properties.append(path + [NUMBER_TYPE])225 append(buffer, '{')226 append(buffer, QUOTED_NUMBER_TYPE)227 append(buffer, float2json(time.mktime(value.timetuple())))228 append(buffer, '}')229 elif _type is Date:230 if NUMBER_TYPE not in sub_schema:231 sub_schema[NUMBER_TYPE] = True232 net_new_properties.append(path + [NUMBER_TYPE])233 append(buffer, '{')234 append(buffer, QUOTED_NUMBER_TYPE)235 append(buffer, float2json(value.unix))236 append(buffer, '}')237 elif _type is timedelta:238 if NUMBER_TYPE not in sub_schema:239 sub_schema[NUMBER_TYPE] = True240 net_new_properties.append(path + [NUMBER_TYPE])241 append(buffer, '{')242 append(buffer, QUOTED_NUMBER_TYPE)243 append(buffer, float2json(value.total_seconds()))244 append(buffer, '}')245 elif _type is Duration:246 if NUMBER_TYPE not in sub_schema:247 sub_schema[NUMBER_TYPE] = True248 net_new_properties.append(path + [NUMBER_TYPE])249 append(buffer, '{')250 append(buffer, QUOTED_NUMBER_TYPE)251 append(buffer, float2json(value.seconds))252 append(buffer, '}')253 elif _type is NullType:254 append(buffer, 'null')255 elif hasattr(value, '__data__'):256 typed_encode(value.__data__(), sub_schema, path, net_new_properties, buffer)257 elif hasattr(value, '__iter__'):258 if NESTED_TYPE not in sub_schema:259 sub_schema[NESTED_TYPE] = {}260 net_new_properties.append(path + [NESTED_TYPE])261 append(buffer, '{')262 append(buffer, QUOTED_NESTED_TYPE)263 _iter2json(value, sub_schema[NESTED_TYPE], path + [NESTED_TYPE], net_new_properties, buffer)264 append(buffer, '}')265 else:266 from mo_logs import Log267 Log.error(text_type(repr(value)) + " is not JSON serializable")268 except Exception as e:269 from mo_logs import Log270 Log.error(text_type(repr(value)) + " is not JSON serializable", cause=e)271def _list2json(value, sub_schema, path, net_new_properties, buffer):272 if not value:273 append(buffer, '[]')274 else:275 sep = '['276 for v in value:277 append(buffer, sep)278 sep = COMMA279 typed_encode(v, sub_schema, path, net_new_properties, buffer)280 append(buffer, ']')281 append(buffer, COMMA)282 append(buffer, QUOTED_EXISTS_TYPE)283 append(buffer, text_type(len(value)))284def _multivalue2json(value, sub_schema, path, net_new_properties, buffer):285 if not value:286 append(buffer, '[]')287 elif len(value) == 1:288 append(buffer, json_encoder(value[0]))289 else:290 sep = '['291 for v in value:292 append(buffer, sep)293 sep = COMMA294 append(buffer, json_encoder(v))295 append(buffer, ']')296def _iter2json(value, sub_schema, path, net_new_properties, buffer):297 append(buffer, '[')298 sep = ''299 count = 0300 for v in value:301 append(buffer, sep)302 sep = COMMA303 typed_encode(v, sub_schema, path, net_new_properties, buffer)304 count += 1305 append(buffer, ']')306 append(buffer, COMMA)307 append(buffer, QUOTED_EXISTS_TYPE)308 append(buffer, text_type(count))309def _dict2json(value, sub_schema, path, net_new_properties, buffer):310 prefix = '{'311 for k, v in sort_using_key(value.items(), lambda r: r[0]):312 if v == None or v == '':313 continue314 append(buffer, prefix)315 prefix = COMMA316 if isinstance(k, binary_type):317 k = utf82unicode(k)318 if not isinstance(k, text_type):319 Log.error("Expecting property name to be a string")320 if k not in sub_schema:321 sub_schema[k] = {}322 net_new_properties.append(path + [k])323 append(buffer, encode_basestring(k))324 append(buffer, COLON)325 typed_encode(v, sub_schema[k], path + [k], net_new_properties, buffer)326 if prefix is COMMA:327 append(buffer, COMMA)328 append(buffer, QUOTED_EXISTS_TYPE)329 append(buffer, '1}')330 else:331 append(buffer, '{')332 append(buffer, QUOTED_EXISTS_TYPE)333 append(buffer, '0}')334TYPE_PREFIX = "~" # u'\u0442\u0443\u0440\u0435-' # "ÑÑÑе"335BOOLEAN_TYPE = TYPE_PREFIX + "b~"336NUMBER_TYPE = TYPE_PREFIX + "n~"337STRING_TYPE = TYPE_PREFIX + "s~"338NESTED_TYPE = TYPE_PREFIX + "N~"339EXISTS_TYPE = TYPE_PREFIX + "e~"340append = UnicodeBuilder.append341QUOTED_BOOLEAN_TYPE = quote(BOOLEAN_TYPE) + COLON342QUOTED_NUMBER_TYPE = quote(NUMBER_TYPE) + COLON343QUOTED_STRING_TYPE = quote(STRING_TYPE) + COLON344QUOTED_NESTED_TYPE = quote(NESTED_TYPE) + COLON345QUOTED_EXISTS_TYPE = quote(EXISTS_TYPE) + COLON346json_type_to_inserter_type = {347 BOOLEAN: BOOLEAN_TYPE,348 INTEGER: NUMBER_TYPE,349 NUMBER: NUMBER_TYPE,350 STRING: STRING_TYPE,351 NESTED: NESTED_TYPE,352 EXISTS: EXISTS_TYPE353}354es_type_to_json_type = {355 "text": "string",356 "string": "string",357 "keyword": "string",358 "float": "number",359 "double": "number",360 "integer": "number",361 "object": "object",362 "nested": "nested",363 "source": "json",364 "boolean": "boolean",365 "exists": "exists"...
parse_types.py
Source:parse_types.py
...39 """40 text = text.lower()41 return text == "1" or text.startswith("y") or text == "true" or text == "on"42@custom_type("QuotedString", r'"(?:[^"\\]|\\.)*"')43def quoted_string_type(text):44 """Custom Parse Type to parse a quoted string.45 Double quotes (") have to be escaped with a46 backslash within the double quotes.47 """48 return text[1:-1]49@custom_type("MathExpression", r"[0-9 +\-*/%.e]+")50def math_expression_type(text):51 """Custom Parse Type which expects a valid mathematical expression52 :param str text: the text which was matched as math expression53 :returns: calculated float number from the math expression54 :rtype: float55 """...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Radish automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
