Python simplejson.load方法代码示例

本文整理汇总了Python中simplejson.load方法的典型用法代码示例。如果您正苦于以下问题:Python simplejson.load方法的具体用法?Python simplejson.load怎么用?Python simplejson.load使用的例子?那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块simplejson的用法示例。

在下文中一共展示了simplejson.load方法的22个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: _load_connectors

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def _load_connectors(self, main_config):
        self.connectors_configs = {}
        if main_config.get("connectors"):
            for connector in main_config['connectors']:
                try:
                    connector_class = TBUtility.check_and_import(connector["type"], self._default_connectors.get(connector["type"], connector.get("class")))
                    self._implemented_connectors[connector["type"]] = connector_class
                    with open(self._config_dir + connector['configuration'], 'r', encoding="UTF-8") as conf_file:
                        connector_conf = load(conf_file)
                        if not self.connectors_configs.get(connector['type']):
                            self.connectors_configs[connector['type']] = []
                        connector_conf["name"] = connector["name"]
                        self.connectors_configs[connector['type']].append({"name": connector["name"], "config": {connector['configuration']: connector_conf}})
                except Exception as e:
                    log.error("Error on loading connector:")
                    log.exception(e)
        else:
            log.error("Connectors - not found! Check your configuration!")
            main_config["remoteConfiguration"] = True
            log.info("Remote configuration is enabled forcibly!") 
开发者ID:thingsboard,项目名称:thingsboard-gateway,代码行数:22,代码来源:tb_gateway_service.py


示例2: __load_iterator_config

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def __load_iterator_config(self):
        if not self.__iterator_file_name:
            if not self.__resolve_iterator_file():
                log.error("[%s] Unable to load iterator configuration from file: file name is not resolved",
                          self.get_name())
                return False
        try:
            iterator_file_path = Path(self.__config_dir + self.__iterator_file_name)
            if not iterator_file_path.exists():
                return False
            with iterator_file_path.open("r") as iterator_file:
                self.__iterator = load(iterator_file)
            log.debug("[%s] Loaded iterator configuration from %s", self.get_name(), self.__iterator_file_name)
        except Exception as e:
            log.error("[%s] Failed to load iterator configuration from %s: %s",
                      self.get_name(), self.__iterator_file_name, str(e))
        return bool(self.__iterator) 
开发者ID:thingsboard,项目名称:thingsboard-gateway,代码行数:22,代码来源:odbc_connector.py


示例3: main

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def main():
    if len(sys.argv) == 1:
        infile = sys.stdin
        outfile = sys.stdout
    elif len(sys.argv) == 2:
        infile = open(sys.argv[1], 'r')
        outfile = sys.stdout
    elif len(sys.argv) == 3:
        infile = open(sys.argv[1], 'r')
        outfile = open(sys.argv[2], 'w')
    else:
        raise SystemExit(sys.argv[0] + " [infile [outfile]]")
    with infile:
        try:
            obj = json.load(infile,
                            object_pairs_hook=json.OrderedDict,
                            use_decimal=True)
        except ValueError:
            raise SystemExit(sys.exc_info()[1])
    with outfile:
        json.dump(obj, outfile, sort_keys=True, indent='    ', use_decimal=True)
        outfile.write('\n') 
开发者ID:gkudos,项目名称:qgis-cartodb,代码行数:24,代码来源:tool.py


示例4: test_object_pairs_hook

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def test_object_pairs_hook(self):
        s = '{"xkd":1, "kcw":2, "art":3, "hxm":4, "qrt":5, "pad":6, "hoy":7}'
        p = [("xkd", 1), ("kcw", 2), ("art", 3), ("hxm", 4),
             ("qrt", 5), ("pad", 6), ("hoy", 7)]
        self.assertEqual(json.loads(s), eval(s))
        self.assertEqual(json.loads(s, object_pairs_hook=lambda x: x), p)
        self.assertEqual(json.load(StringIO(s),
                                   object_pairs_hook=lambda x: x), p)
        od = json.loads(s, object_pairs_hook=OrderedDict)
        self.assertEqual(od, OrderedDict(p))
        self.assertEqual(type(od), OrderedDict)
        # the object_pairs_hook takes priority over the object_hook
        self.assertEqual(json.loads(s,
                                    object_pairs_hook=OrderedDict,
                                    object_hook=lambda x: None),
                         OrderedDict(p)) 
开发者ID:gkudos,项目名称:qgis-cartodb,代码行数:18,代码来源:test_decode.py


示例5: __init__

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def __init__(self, http_error):
    """Create a ServerRequestException from a given urllib2.HTTPError.
    Args:
      http_error: The HTTPError that the ServerRequestException will be
        based on.
    """
    error_details = None
    if http_error.fp:
      try:
        error_body = json.load(http_error.fp)
        error_details = ['%s: %s' % (detail['message'], detail['debug_info'])
                         for detail in error_body['error']['errors']]
      except (ValueError, TypeError, KeyError):
        pass
    if error_details:
      error_message = ('HTTP %s (%s) error when communicating with URL: %s.  '
                       'Details: %s' % (http_error.code, http_error.reason,
                                        http_error.filename, error_details))
    else:
      error_message = ('HTTP %s (%s) error when communicating with URL: %s.' %
                       (http_error.code, http_error.reason,
                        http_error.filename))
    super(ServerRequestException, self).__init__(error_message) 
开发者ID:elsigh,项目名称:browserscope,代码行数:26,代码来源:endpointscfg.py


示例6: add_async_request

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def add_async_request(self, method, relative_url, headers, body, source_ip,
                        server_name=None, version=None, instance_id=None):
    """Dispatch an HTTP request asynchronously.
    Args:
      method: A str containing the HTTP method of the request.
      relative_url: A str containing path and query string of the request.
      headers: A list of (key, value) tuples where key and value are both str.
      body: A str containing the request body.
      source_ip: The source ip address for the request.
      server_name: An optional str containing the server name to service this
          request. If unset, the request will be dispatched to the default
          server.
      version: An optional str containing the version to service this request.
          If unset, the request will be dispatched to the default version.
      instance_id: An optional str containing the instance_id of the instance to
          service this request. If unset, the request will be dispatched to
          according to the load-balancing for the server and version.
    """
    fake_socket = FakeRequestSocket(method, relative_url, headers, body)
    self._server.AddEvent(0, lambda: (fake_socket, (source_ip, self._port))) 
开发者ID:elsigh,项目名称:browserscope,代码行数:23,代码来源:dev_appserver.py


示例7: main

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def main():
    if len(sys.argv) == 1:
        infile = sys.stdin
        outfile = sys.stdout
    elif len(sys.argv) == 2:
        infile = open(sys.argv[1], 'rb')
        outfile = sys.stdout
    elif len(sys.argv) == 3:
        infile = open(sys.argv[1], 'rb')
        outfile = open(sys.argv[2], 'wb')
    else:
        raise SystemExit(sys.argv[0] + " [infile [outfile]]")
    try:
        obj = json.load(infile,
                        object_pairs_hook=json.OrderedDict,
                        use_decimal=True)
    except ValueError, e:
        raise SystemExit(e) 
开发者ID:linuxscout,项目名称:mishkal,代码行数:20,代码来源:tool.py


示例8: _try_load_json

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def _try_load_json(self, jsonfile):
        logger.debug("Loading json configuration from %s" % jsonfile)
        fd = None
        try:
            fd = open(jsonfile)
        except Exception as e:
            logger.debug(
                "Error reading configuration file %s: %s; "
                "ignoring..." % (jsonfile, e)
            )
            return {}
        config = None
        try:
            config = json.load(fd)
        except Exception as e:
            logger.error(
                "Error parsing configuration file %s: %s" % (jsonfile, e)
            )
            return {}
        logger.trace("contents read from %s: %s" % (jsonfile, config))
        return config 
开发者ID:mendix,项目名称:cf-mendix-buildpack,代码行数:25,代码来源:config.py


示例9: load_config

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def load_config(yaml_file):
    logger.debug("Loading configuration from %s" % yaml_file)
    fd = None
    try:
        fd = open(yaml_file)
    except Exception as e:
        logger.error(
            "Error reading configuration file %s, ignoring..." % yaml_file
        )
        return
    try:
        return yaml.load(fd, Loader=yaml.FullLoader)
    except Exception as e:
        logger.error(
            "Error parsing configuration file %s: %s" % (yaml_file, e)
        )
        return 
开发者ID:mendix,项目名称:cf-mendix-buildpack,代码行数:20,代码来源:config.py


示例10: get_host_info

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def get_host_info(self):
        ''' Get variables about a specific host '''
        if len(self.index) == 0:
            # Need to load index from cache
            self.load_index_from_cache()
        if self.args.host not in self.index:
            # try updating the cache
            self.do_api_calls_update_cache()
            if self.args.host not in self.index:
                # host might not exist anymore
                return self.json_format_dict({}, True)
        (region, instance_id) = self.index[self.args.host]
        instance = self.get_instance(region, instance_id)
        return self.json_format_dict(self.get_host_info_dict_from_instance(instance), True) 
开发者ID:haproxytech,项目名称:cloud-blueprints,代码行数:20,代码来源:ec2.py


示例11: load

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def load(self):
        con = sqlite3.connect(self.tmp_cookie_file)
        cur = con.cursor()
        cur.execute('select host, path, isSecure, expiry, name, value from moz_cookies '
                    'where host like "%{}%"'.format(self.domain_name))
        cj = http.cookiejar.CookieJar()
        for item in cur.fetchall():
            c = create_cookie(*item)
            cj.set_cookie(c)
        con.close()
        self.__add_session_cookies(cj)
        self.__add_session_cookies_lz4(cj)
        return cj 
开发者ID:borisbabic,项目名称:browser_cookie3,代码行数:18,代码来源:__init__.py


示例12: read_config

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def read_config(config_file):
    config_file = os.path.abspath(os.path.expanduser(config_file))
    if not os.path.isfile(config_file):
        print_err("Config file not found!")
        sys.exit(1)
    with open(config_file, "r") as f:
        config = json.load(f)
        if "session" not in config:
            config["session"] = "SESSION"
        if "fuzzer" not in config:
            config["fuzzer"] = "afl-fuzz"
        return config 
开发者ID:rc0r,项目名称:afl-utils,代码行数:19,代码来源:afl_multicore.py


示例13: load

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def load(source_dir: Union[os.PathLike, str]) -> Any:
    """
    Load an object from a directory, saved by
    ``gordo.serializer.pipeline_serializer.dump``
    This take a directory, which is either top-level, meaning it contains
    a sub directory in the naming scheme: "n_step=<int>-class=<path.to.Class>"
    or the aforementioned naming scheme directory directly. Will return that
    unsterilized object.
    Parameters
    ----------
    source_dir: Union[os.PathLike, str]
        Location of the top level dir the pipeline was saved
    Returns
    -------
    Union[GordoBase, Pipeline, BaseEstimator]
    """
    # This source dir should have a single pipeline entry directory.
    # may have been passed a top level dir, containing such an entry:
    with open(os.path.join(source_dir, "model.pkl"), "rb") as f:
        return pickle.load(f) 
开发者ID:equinor,项目名称:gordo,代码行数:26,代码来源:serializer.py


示例14: load

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def load(self):
        """
        Extract tabular data as |TableData| instances from a dict object.
        |load_source_desc_text|
        :rtype: |TableData| iterator
        .. seealso::
            :py:meth:`.JsonTableFileLoader.load()`
        """
        self._validate()
        self._logger.logging_load()
        formatter = JsonTableFormatter(self.source)
        formatter.accept(self)
        return formatter.to_table_data() 
开发者ID:thombashi,项目名称:pytablereader,代码行数:21,代码来源:core.py


示例15: main

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def main():
    if len(sys.argv) == 1:
        infile = sys.stdin
        outfile = sys.stdout
    elif len(sys.argv) == 2:
        infile = open(sys.argv[1], 'rb')
        outfile = sys.stdout
    elif len(sys.argv) == 3:
        infile = open(sys.argv[1], 'rb')
        outfile = open(sys.argv[2], 'wb')
    else:
        raise SystemExit(sys.argv[0] + " [infile [outfile]]")
    try:
        obj = simplejson.load(infile)
    except ValueError, e:
        raise SystemExit(e) 
开发者ID:tvalacarta,项目名称:tvalacarta,代码行数:18,代码来源:tool.py


示例16: setUpClass

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def setUpClass(cls):
        swagger_file = os.path.join(SAMPLES_PATH, cls.swagger_filename)
        with codecs.open(swagger_file, 'r', encoding='utf-8') as _file:
            doc = json.load(_file)
        if doc is None:
            raise SkipTest('File is empty')
        if cls.example_filename:
            example_file = os.path.join(SAMPLES_PATH, cls.example_filename)
            with codecs.open(example_file, 'r', encoding='utf-8') as _file:
                cls.examples = json.load(_file)
            if not cls.examples:
                raise SkipTest('Example file is empty')
        cls.swagger_doc = BaseSwaggerObject(doc)
        cls.exampilator = cls.swagger_doc.exampilator 
开发者ID:Arello-Mobile,项目名称:swagger2rst,代码行数:21,代码来源:test.py


示例17: prepare_env

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def prepare_env(cnt, file_name=True, inline=False):
        this = {}
        if file_name:
            this['file_name_json'] = os.path.join(SAMPLES_PATH, '{}.json'.format(cnt))
            this['file_name_rst'] = os.path.join(SAMPLES_PATH,
                                                 '{}{inline}.rst'.format(cnt, inline='_inline' if inline else '')
                                                 )
            with codecs.open(this['file_name_json'], 'r', encoding='utf-8') as _file:
                doc = json.load(_file)
        else:
            this['file_name_json'] = False
            this['file_name_rst'] = False
            doc = json.load(cnt)
        this['swagger_doc'] = rst.SwaggerObject(doc)
        doc_module = importlib.import_module('swg2rst.utils.rst')
        jinja_env = Environment(lstrip_blocks=True, trim_blocks=True)
        jinja_env.loader = PackageLoader('swg2rst')
        for name, function in inspect.getmembers(doc_module, inspect.isfunction):
            jinja_env.filters[name] = function
        jinja_env.filters['sorted'] = sorted
        template = jinja_env.get_template('main.rst')
        this['raw_rst'] = template.render(doc=this['swagger_doc'], inline=inline)
        this['pattern'] = re.compile(r'[idm]_\w{32}')
        this['normalize'] = lambda x: x[:-1] if x[-1] == '\n' else x
        return this 
开发者ID:Arello-Mobile,项目名称:swagger2rst,代码行数:27,代码来源:test.py


示例18: get_database_JSON

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def get_database_JSON(self):
        """
        Load existing JSON representation of import database (created last time OWL ontology was saved)
        Will be updated if database has changed.
        """
        with open(self.database_path) as data_file:
            dbObject = json.load(data_file, object_pairs_hook=OrderedDict)
        with open(self.product_type_path) as data_file:
            dbObject2 = json.load(data_file, object_pairs_hook=OrderedDict)
        for item in dbObject2['index']:
            dbObject['index'][item] = dbObject2['index'][item]
        self.database = dbObject
        # Create a reverse-lookup index by food source label, or extract, concentrate etc.
        for item in dbObject['index']:
            # C0228 == extract, concentrate or isolate of plant or animal
            if item[0] == 'B' or self.itemAncestor(item, ['C0228']):
                entity = dbObject['index'][item]
                if not (entity['status'] == 'deprecated' or entity['status'] == 'ignore'):
                    self.label_reverse_lookup[entity['label']['value'].lower()] = entity 
开发者ID:FoodOntology,项目名称:foodon,代码行数:26,代码来源:subset.py


示例19: get_host_info

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def get_host_info(self):
        ''' Get variables about a specific host '''
        if len(self.index) == 0:
            # Need to load index from cache
            self.load_index_from_cache()
        if not self.args.host in self.index:
            # try updating the cache
            self.do_api_calls_update_cache()
            if not self.args.host in self.index:
                # host might not exist anymore
                return self.json_format_dict({}, True)
        (region, instance_id) = self.index[self.args.host]
        instance = self.get_instance(region, instance_id)
        return self.json_format_dict(self.get_host_info_dict_from_instance(instance), True) 
开发者ID:scottslowe,项目名称:learning-tools,代码行数:20,代码来源:ec2.py


示例20: _parse_and_compare

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def _parse_and_compare(self, domain_name, data, expected_results, whois_entry=WhoisEntry):
        results = whois_entry.load(domain_name, data)
        fail = 0
        total = 0
        # Compare each key
        for key in expected_results:
            total += 1
            result = results.get(key)
            if isinstance(result, datetime.datetime):
                result = str(result)
            expected = expected_results.get(key)
            if expected != result:
                print("%s \t(%s):\t %s != %s" % (domain_name, key, result, expected))
                fail += 1
        if fail:
            self.fail("%d/%d sample whois attributes were not parsed properly!"
                      % (fail, total)) 
开发者ID:richardpenman,项目名称:whois,代码行数:19,代码来源:test_parser.py


示例21: _create_connector

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def _create_connector(self,
                          config_file_name,
                          send_on_change=OdbcConnector.DEFAULT_SEND_IF_CHANGED,
                          unknown_rpc=OdbcConnector.DEFAULT_ENABLE_UNKNOWN_RPC,
                          override_rpc_params=OdbcConnector.DEFAULT_OVERRIDE_RPC_PARAMS,
                          connection_str="",
                          reconnect=OdbcConnector.DEFAULT_RECONNECT_STATE):
        with open(self.CONFIG_PATH + config_file_name, 'r', encoding="UTF-8") as file:
            self.config = load(file)
            if "sendDataOnlyOnChange" not in self.config["mapping"]:
                self.config["mapping"]["sendDataOnlyOnChange"] = send_on_change
            if "serverSideRpc" not in self.config:
                self.config["serverSideRpc"] = {}
            if "enableUnknownRpc" not in self.config["serverSideRpc"]:
                self.config["serverSideRpc"]["enableUnknownRpc"] = unknown_rpc
            if "overrideRpcConfig" not in self.config["serverSideRpc"]:
                self.config["serverSideRpc"]["overrideRpcConfig"] = override_rpc_params
            self.config["connection"]["reconnect"] = reconnect
            if connection_str:
                self.config["connection"]["str"] = connection_str
            else:
                self.config["connection"]["str"] = self.db_connection_str
                self._init_test_db()
            self.connector = OdbcConnector(self.gateway, self.config, "odbc")
            self.connector.open()
            sleep(1)  # some time to init 
开发者ID:thingsboard,项目名称:thingsboard-gateway,代码行数:35,代码来源:test_odbc_connector.py


示例22: _create_connector

# 需要导入模块: import simplejson [as 别名]
# 或者: from simplejson import load [as 别名]
def _create_connector(self, config_file_name):
        with open(self.CONFIG_PATH + config_file_name, 'r', encoding="UTF-8") as file:
            self.config = load(file)
            self.connector = CanConnector(self.gateway, self.config, "can")
            self.connector.open()
            sleep(1)  # some time to init 
开发者ID:thingsboard,项目名称:thingsboard-gateway,代码行数:8,代码来源:test_can_connector.py



注:本文中的simplejson.load方法示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。