我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用jsonpickle.encode()。
def _construct_report_request(self): """Construct a report request.""" report = None with self._mutex: report = ttypes.ReportRequest(self._runtime, self._span_records, None) self._span_records = [] for span in report.span_records: for log in span.log_records: index = span.log_records.index(log) if log.payload_json is not None: try: log.payload_json = \ jsonpickle.encode(log.payload_json, unpicklable=False, make_refs=False, max_depth=constants.JSON_MAX_DEPTH) except: log.payload_json = jsonpickle.encode(constants.JSON_FAIL) span.log_records[index] = log return report
def form_encode_parameters(form_parameters): """Form encodes a dictionary of form parameters Args: form_parameters (dictionary): The given dictionary which has atleast one model to form encode. Returns: dict: A dictionary of form encoded properties of the model. """ encoded = dict() for key, value in form_parameters.items(): encoded.update(APIHelper.form_encode(value, key)) return encoded
def form_encode_parameters(form_parameters, array_serialization="indexed"): """Form encodes a dictionary of form parameters Args: form_parameters (dictionary): The given dictionary which has atleast one model to form encode. array_serialization (str): The format of array parameter serialization. Returns: dict: A dictionary of form encoded properties of the model. """ encoded = [] for key, value in form_parameters.items(): encoded += APIHelper.form_encode(value, key, array_serialization) return encoded
def handle_message(customer_uuid): """Handles messages from a phone number with body @param customer_uuid The UUID of the sender @data message_body The body of the message """ data = jsonpickle.decode(request.data.decode("utf-8")) message_body = data["message_body"] # Retrieve the message and create the customer customer = Customer.load_from_db(customer_uuid) messaging.on_message_recieve(customer, message_body) return jsonpickle.encode(dict( success=True ))
def handle_sms(): """Handles Twillio SMS message""" customer_phone_number = request.values["From"] text_message_body = request.values["Body"] # Retrieve the customer from the DB or create a new one customer = Customer.get_customer_by_phone_number(customer_phone_number) if not customer: # Save the customer to the DB customer = Customer.create_new({ CFields.PHONE_NUMBER: customer_phone_number, }) customer.create() customer_lc.on_new_customer() messaging.on_message_recieve(customer, text_message_body) return jsonpickle.encode(dict( success=True ))
def ToJson(object): returnValue = "{}" try: returnValue = encode(object, unpicklable=False, make_refs=False) except: exception('ToJson Failed') return returnValue #{ # "id":"some_unique_id", # "type":"date/interval", # "unit":"minute/hour/day/week/month/year", # "interval":1, # "weekday":"monday/tuesday/...", # "start_date":"date and hour", # "end_date":"date and hour", # "title":"sometext", # "notify":"yes/no", # "actions": # [ # ] #} #AccountId = None
def Search(self): cells = Cell.all('wlan0') self.endpoints = list(cells) serializableEndpoints = [] for i in self.endpoints: wifiEndpoint = WifiEndpoint() wifiEndpoint.ssid = i.ssid wifiEndpoint.signal = i.signal wifiEndpoint.quality = i.quality wifiEndpoint.frequency = i.frequency wifiEndpoint.bitrates = i.bitrates wifiEndpoint.encrypted = i.encrypted wifiEndpoint.channel = i.channel wifiEndpoint.address = i.address wifiEndpoint.mode = i.mode serializableEndpoints.append(wifiEndpoint) #print (i.ssid, i.signal, i.quality, i.frequency) schemes = Scheme.all() #print(jsonpickle.encode(schemes)) return jsonpickle.encode(serializableEndpoints)
def on_websocket_message(self, restoreFile): #fileinfo = self.request.files['ZYNTHIAN_RESTORE_FILE'][0] #restoreFile = fileinfo['filename'] with open(restoreFile , "rb") as f: validRestoreItems = get_backup_items(SYSTEM_BACKUP_ITEMS_FILE) validRestoreItems += get_backup_items(DATA_BACKUP_ITEMS_FILE) with zipfile.ZipFile(f,'r') as restoreZip: for member in restoreZip.namelist(): if self.is_valid_restore_item(validRestoreItems, member): logMessage = "restored: " + member restoreZip.extract(member, "/") logging.debug(logMessage) message = ZynthianWebSocketMessage('RestoreMessageHandler', logMessage) self.websocket.write_message(jsonpickle.encode(message)) else: logging.warn("restore of " + member + " not permitted") restoreZip.close() f.close() os.remove(restoreFile) message = ZynthianWebSocketMessage('RestoreMessageHandler', 'EOCOMMAND') self.websocket.write_message(jsonpickle.encode(message));
def save(self, filepath): """Saves the optimizer parameters to the specifiec path as JSON. Parameters ---------- filepath: str The file path. """ # check and create dirs if not os.path.exists(os.path.dirname(filepath)): subdirs = os.path.dirname(filepath) if subdirs is not None and subdirs != '': os.makedirs(subdirs) with open(filepath, 'wb') as f: json = jsonpickle.encode(self) f.write(json)
def save(self, filepath): """Saves the model parameters to the specifiec path as JSON. Parameters ---------- filepath: str The file path. """ # check and create dirs if not os.path.exists(os.path.dirname(filepath)): subdirs = os.path.dirname(filepath) if subdirs is not None and subdirs != '': os.makedirs(subdirs) with open(filepath, 'wb') as f: json = jsonpickle.encode(self) f.write(json)
def _get_exported_string(format_stripped: str, deck: Deck) -> str: """ Helper function to Deck.export() :param format_stripped: Desired format stripped of any spaces and lowercase :type format_stripped: str :param deck: instance of a Deck :type deck: :ref:`Deck` :return: YAML/JSON string of the deck :rtype: str :raises UnknownFormat: when it doesn't recognize format_stripped """ if format_stripped == "yaml" or format_stripped == "yml": exported = yaml.dump(deck) log.debug("Exported deck %r to a yaml string", deck) elif format_stripped == "json": exported = jsonpickle.encode(deck) log.debug("Exported deck %r to a json string", deck) else: log.debug("Unknown format: %s", format) raise UnknownFormat("Unknown format: {}".format(format)) return exported
def save_game(save_path, players, relations, market, provinces): state = dict() jsonpickle.set_preferred_backend('simplejson') #save = shelve.open(file_name, flag = "n", writeback = False) for p, player in players.items(): state[p] = player for re, rel in relations.items(): state[re] = rel #state["relations"] = relations for p, prov in provinces.items(): state[prov.name] = prov #for uc, unciv in uncivilized_minors.items(): # state[uc] = unciv state["market"] = market with open(save_path, 'w') as save: save.write(jsonpickle.encode(state, keys = True, warn = True)) save.close() #save.close() #print("%s saved to disk \n" % (file_name))
def find_peaks_automatically_and_export (self, n_expected=None, include_headers=False) : err = StringIO() peak_times, bandwidth = self.find_consensus_peaks( n_expected=n_expected, err=err) table, errors = self.extract_peak_areas(peak_times, bandwidth, err=err) if include_headers : table.insert(0, ["Sample ID"] + ["Peak %d" % (i+1) for i in range(len(peak_times)) ]) table.insert(1, [None] + ["%.4fm" % x for x in peak_times]) return { "data_type" : "gc_ms", "auto_peak" : True, "bandwidth" : bandwidth, "bandwidth_auto" : True, #bandwidth_auto, "peak_times" : peak_times, "sample_data" : table, "errors" : errors, "samples" : jsonpickle.encode(self.samples), }
def find_peaks_by_range_and_export (self, rt_ranges, molecule_names=None) : assert (molecule_names is None) or (len(molecule_names) == len(rt_ranges)) table, errors = self.extract_peak_areas_by_range(rt_ranges) peak_ranges = ["%.4f - %.4fm" %(x,y) for (x,y) in rt_ranges] if (molecule_names is not None) : table.insert(0, ["Sample ID"] + list(molecule_names)) table.insert(1, [None] + peak_ranges) return { "data_type" : "gc_ms", "auto_peak" : False, "peak_times" : peak_ranges, "sample_data" : table, "errors" : errors, "samples" : jsonpickle.encode(self.samples), }
def getGPUUsage(): try: pynvml.nvmlInit() count = pynvml.nvmlDeviceGetCount() if count == 0: return None result = {"driver": pynvml.nvmlSystemGetDriverVersion(), "gpu_count": int(count) } i = 0 gpuData = [] while i<count: handle = pynvml.nvmlDeviceGetHandleByIndex(i) mem = pynvml.nvmlDeviceGetMemoryInfo(handle) gpuData.append({"device_num": i, "name": pynvml.nvmlDeviceGetName(handle), "total": round(float(mem.total)/1000000000, 2), "used": round(float(mem.used)/1000000000, 2)}) i = i+1 result["devices"] = jsonpickle.encode(gpuData, unpicklable=False) except Exception as e: result = {"driver": "No GPU!", "gpu_count": 0, "devices": []} return result
def predict(imagepath, target_x, target_y, name, model): if imagepath.startswith('http://') or imagepath.startswith('https://') or imagepath.startswith('ftp://'): response = requests.get(imagepath) img = Image.open(BytesIO(response.content)) img = img.resize((target_x, target_y)) else: if not os.path.exists(imagepath): raise Exception('Input image file does not exist') img = image.load_img(imagepath, target_size=(target_x, target_y)) x = image.img_to_array(img) x = np.expand_dims(x, axis=0) x = processInputImage(name, x) preds = decodePrediction(name, model.predict(x)) result = [] for p in preds[0]: result.append({"synset": p[0], "text": p[1], "prediction": float("{0:.2f}".format((p[2] * 100)))}) return json.loads(jsonpickle.encode(result, unpicklable=False))
def predictint(): message = "Success" code = 200 result = [] try: start = datetime.utcnow() data = request.args.get('data') result = intentanalyzer.predict(data) result = json.loads(jsonpickle.encode(result, unpicklable=False)) logmgr.LogPredSuccess("intent", constants.ServiceTypes.LangIntent, start) except Exception as e: code = 500 message = str(e) logmgr.LogPredError("intent", constants.ServiceTypes.LangIntent, start, message) return jsonify({"statuscode": code, "message": message, "result": result})
def tokenize(data, language="english", filterStopWords = False, tagging = False): result = {} tags = [] filterChars = [",", ".", "?", ";", ":", "'", "!", "@", "#", "$", "%", "&", "*", "(", ")", "+", "{", "}", "[", "]", "\\", "|"] sent_token = nltk.tokenize.sent_tokenize(data, language) word_token = nltk.tokenize.word_tokenize(data, language) word_token = [w for w in word_token if not w in filterChars] if filterStopWords is True: stop_words = set(stopwords.words(language)) word_token = [w for w in word_token if not w in stop_words] if tagging is True: tags = nltk.pos_tag(word_token) result = {"sent_token": sent_token, "word_token": word_token, "pos_tag": tags} return json.loads(jsonpickle.encode(result, unpicklable=False))
def main(p, folder): save_ = BASE_PATH + folder while p.poll() is None: result = await process_frame() if result: file_name = 'frame_' + str(result[0]) + '.txt' with open(save_ + file_name, 'w') as f: for res in result[1]: f.write(jsonpickle.encode(res) + "\n") remaining = Q.qsize() print(str(remaining) + " frames left") for i in range(remaining): result = await process_frame() if result: print(str(Q.qsize()) + " left in the queue") file_name = 'frame_' + str(result[0]) + '.txt' with open(save_ + file_name, 'w') as f: for res in result[1]: f.write(jsonpickle.encode(res) + "\n")
def test_read_settings(mock_open): mock_open.return_value = StringIO(encode(default_settings)) assert read_settings() == default_settings broken_settings = deepcopy(default_settings) broken_settings["General"] = "random string" # simulate user messing with settings mock_open.return_value = StringIO(encode(broken_settings)) assert read_settings() == default_settings mock_open.side_effect = FileNotFoundError("mock settings file not existing") assert read_settings() == default_settings mock_open.side_effect = JSONDecodeError( "mock settings file not being decodable - someone messed with the file", encode(default_settings), 10 # just a random value ) assert read_settings() == default_settings
def invoke_api(ip, command, logger, headers=None): if headers is None: headers = {} try: method = command.get_method() url = "https://{0}:9000{1}".format(ip, command.get_url()) data = jsonpickle.encode(command, unpicklable=False) if "get" == method.lower(): response = requests.get(url=url, headers=headers, verify=False) else: headers["Content-Type"] = "application/json" response = requests.request(method=method, data=str(data), url=url, headers=headers, verify=False) json_obj = validate_response(response) return command.process_response(json_obj) except Exception as e: logger.error("Failed to execute command: %s", e) return None
def set_aad_cache(token, cache): """Set AAD token cache.""" set_config_value('aad_token', jsonpickle.encode(token)) set_config_value('aad_cache', jsonpickle.encode(cache))
def json_dumps(value): return jsonpickle.encode(value)
def fingerprint(obj): serialized = json_dumps(obj) with open('/tmp/v1.json', 'w') as f: print('see /tmp/v1.json') f.write(serialized) hasher = hashlib.sha1() try: hasher.update(serialized) except TypeError: hasher.update(serialized.encode('utf-8')) return hasher.hexdigest()
def csv_reader(filename, encoding='utf-8'): with io.open(filename, 'r', encoding=encoding) as f: if sys.version_info >= (3, 0): yield csv.reader(f) else: yield csv.reader(line.encode(encoding, 'ignore') for line in f)
def _make_flow(request, scopes, return_url=None): """Creates a Web Server Flow Args: request: A Django request object. scopes: the request oauth2 scopes. return_url: The URL to return to after the flow is complete. Defaults to the path of the current request. Returns: An OAuth2 flow object that has been stored in the session. """ # Generate a CSRF token to prevent malicious requests. csrf_token = hashlib.sha256(os.urandom(1024)).hexdigest() request.session[_CSRF_KEY] = csrf_token state = json.dumps({ 'csrf_token': csrf_token, 'return_url': return_url, }) flow = client.OAuth2WebServerFlow( client_id=django_util.oauth2_settings.client_id, client_secret=django_util.oauth2_settings.client_secret, scope=scopes, state=state, redirect_uri=request.build_absolute_uri( urlresolvers.reverse("google_oauth:callback"))) flow_key = _FLOW_KEY.format(csrf_token) request.session[flow_key] = jsonpickle.encode(flow) return flow
def get_prep_value(self, value): """Overrides ``models.Field`` method. This is used to convert the value from an instances of this class to bytes that can be inserted into the database. """ if value is None: return None else: return encoding.smart_text( base64.b64encode(jsonpickle.encode(value).encode()))
def json_serialize(obj): """JSON Serialization of a given object. Args: obj (object): The object to serialise. Returns: str: The JSON serialized string of the object. """ if obj is None: return None # Resolve any Names if it's one of our objects that needs to have this called on if isinstance(obj, list): value = list() for item in obj: try: value.append(item.resolve_names()) except (AttributeError, TypeError): value.append(item) obj = value else: try: obj = obj.resolve_names() except (AttributeError, TypeError): obj = obj return jsonpickle.encode(obj, False)
def form_encode(obj, instanceName): """Encodes a model in a form-encoded manner such as person[Name] Args: obj (object): The given Object to form encode. instanceName (string): The base name to appear before each entry for this object. Returns: dict: A dictionary of form encoded properties of the model. """ retval = dict() # If we received an object, resolve it's field names. value = APIHelper.resolve_name(obj) if value is None: return None elif isinstance(value, list): for index, entry in enumerate(value): retval.update(APIHelper.form_encode(entry, instanceName + "[" + str(index) + "]")) elif isinstance(value, dict): for item in value: retval.update(APIHelper.form_encode(value[item], instanceName + "[" + item + "]")) else: retval[instanceName] = obj return retval
def get_record(self): web_show = WebShow(self.playrecords) # return jsonpickle.encode(web_show, unpicklable=False) return web_show #????????
def to_json(bundle): return jsonpickle.encode(bundle)
def form_encode(obj, instanceName): """Encodes a model in a form-encoded manner such as person[Name] Args: obj (object): The given Object to form encode. instanceName (string): The base name to appear before each entry for this object. Returns: dict: A dictionary of form encoded properties of the model. """ # Resolve the names first value = APIHelper.resolve_name(obj) retval = dict() if value is None: return None # Loop through every item we need to send for item in value: if isinstance(value[item], list): # Loop through each item in the list and add it by number i = 0 for entry in value[item]: retval.update(APIHelper.form_encode(entry, instanceName + "[" + item + "][" + str(i) + "]")) i += 1 elif isinstance(value[item], dict): # Loop through each item in the dictionary and add it retval.update(APIHelper.form_encode(value[item], instanceName + "[" + item + "]")) else: # Add the current item retval[instanceName + "[" + item + "]"] = value[item] return retval
def process_bind_param(self, value, engine): return unicode(jsonpickle.encode(value))
def get_state(self): return jsonpickle.encode({ 'start_date': self._start_date, 'static_unit_net_value': self._static_unit_net_value, 'units': self._units, 'accounts': { name: account.get_state() for name, account in six.iteritems(self._accounts) } }).encode('utf-8')
def json_serialize(obj): """JSON Serialization of a given object. Args: obj (object): The object to serialise. Returns: str: The JSON serialized string of the object. """ if obj is None: return None # Resolve any Names if it's one of our objects that needs to have this called on if isinstance(obj, list): value = list() for item in obj: if hasattr(item, "_names"): value.append(APIHelper.to_dictionary(item)) else: value.append(item) obj = value else: if hasattr(obj, "_names"): obj = APIHelper.to_dictionary(obj) return jsonpickle.encode(obj, False)
def form_encode(obj, instance_name, array_serialization="indexed"): """Encodes a model in a form-encoded manner such as person[Name] Args: obj (object): The given Object to form encode. instance_name (string): The base name to appear before each entry for this object. array_serialization (string): The format of array parameter serialization. Returns: dict: A dictionary of form encoded properties of the model. """ retval = [] # If we received an object, resolve it's field names. if hasattr(obj, "_names"): obj = APIHelper.to_dictionary(obj) if obj is None: return [] elif isinstance(obj, list): for element in APIHelper.serialize_array(instance_name, obj, array_serialization): retval += APIHelper.form_encode(element[1], element[0], array_serialization) elif isinstance(obj, dict): for item in obj: retval += APIHelper.form_encode(obj[item], instance_name + "[" + item + "]", array_serialization) else: retval.append((instance_name, obj)) return retval
def toJson(obj): json_obj = jsonpickle.encode(obj, unpicklable=False) return json_obj
def setUp(self): self.fake_model = tests_models.CredentialsModel() self.fake_model_field = self.fake_model._meta.get_field('credentials') self.field = models.CredentialsField(null=True) self.credentials = client.Credentials() self.pickle_str = _helpers._from_bytes( base64.b64encode(pickle.dumps(self.credentials))) self.jsonpickle_str = _helpers._from_bytes( base64.b64encode(jsonpickle.encode(self.credentials).encode()))
def save(self) -> None: with open(self.dump_filename, "w", encoding='utf-8') as f: f.write(jsonpickle.encode(self, f))
def to_json(self): """ :return: ???????????? ? json. """ return jsonpickle.encode(self)
def json_serialize(cls, obj): """ JSON Serialization of a given object. Args: obj (object): The object to serialise. Returns: str: The JSON serialized string of the object. """ if obj is None: return None # Resolve any Names if it's one of our objects # that needs to have this called on if isinstance(obj, list): value = list() for item in obj: try: value.append(item.resolve_names()) except (AttributeError, TypeError): value.append(item) obj = value else: try: obj = obj.resolve_names() except (AttributeError, TypeError): obj = obj return jsonpickle.encode(obj, False)
def form_encode(cls, obj, instanceName): """ Encodes a model in a form-encoded manner such as person[Name] Args: obj (object): The given Object to form encode. instanceName (string): The base name to appear before each entry for this object. Returns: dict: A dictionary of form encoded properties of the model. """ # Resolve the names first value = APIHelper.resolve_name(obj) retval = dict() if value is None: return None # Loop through every item we need to send for item in value: if isinstance(value[item], list): # Loop through each item in the list and add it by number i = 0 for entry in value[item]: retval.update(APIHelper.form_encode( entry, instanceName + "[" + item + "][" + str( i) + "]")) i += 1 elif isinstance(value[item], dict): # Loop through each item in the dictionary and add it retval.update(APIHelper.form_encode(value[item], instanceName + "[" + item + "]")) else: # Add the current item retval[instanceName + "[" + item + "]"] = value[item] return retval
def json_of_spec(spec): return json.loads(jsonpickle.encode(spec, unpicklable=False))
