Python bson.json_util 模块，loads() 实例源码

我们从Python开源项目中，提取了以下42个代码示例，用于说明如何使用bson.json_util.loads()。

项目：xgovctf 作者：alphagov | 项目源码 | 文件源码

def insert_problem_from_json(blob):
    """
    Converts json blob of problem(s) into dicts. Runs insert_problem on each one.
    See insert_problem for more information.

    Returns:
        A list of the created problem pids if an array of problems is specified.
    """

    result = json_util.loads(blob)

    if type(result) == list:
        return [insert_problem(problem) for problem in result]
    elif type(result) == dict:
        return insert_problem(result)
    else:
        raise InternalException("JSON blob does not appear to be a list of problems or a single problem.")

项目：dataIO-project 作者：MacHu-GWU | 项目源码 | 文件源码

def test_all(self):
        data = {
            "int": 100,
            "float": 3.1415926535,
            "str": "string example ?????",
            "bytes": "bytes example ?????".encode("utf-8"),
            "boolean": True,
            "datetime": datetime.now()
        }
        js = json_util.dumps(data)

        data1 = json_util.loads(js)
        self.assertEqual(data["int"], data1["int"])
        self.assertAlmostEqual(data["float"], data1["float"], delta=0.0001)
        self.assertEqual(data["str"], data1["str"])
        self.assertEqual(data["boolean"], data1["boolean"])

        print(data1["bytes"])
        print(data1["datetime"])
        print(json_util.dumps(data, sort_keys=True, indent=4))


#--- Unittest ---

项目：pypers 作者：frankosan | 项目源码 | 文件源码

def verify_token(username, token):
    """
    Verify validity of token
    """
    s = TimedJWSSerializer(app.config['SECRET_KEY'])

    try:
        ut.pretty_print("Trying to load the token")
        data = s.loads(token)
    except SignatureExpired:
        ut.pretty_print("ERROR: Expired Token")
        return False
    except BadSignature:
        ut.pretty_print("ERROR: Invalid Token")
        return False
    else:
        ut.pretty_print("Token successfully loaded")
        stored = db.sessions.find_one(filter={'username': data['username']}, sort=[('_id',-1)])

        if not stored:
            return False
        result = json_util.loads(json_util.dumps(stored))

        return pwd_context.verify(data['password'], result['password_hash']) and data['username'] == username

项目：picoCTF 作者：picoCTF | 项目源码 | 文件源码

def get_json_objects(files):
    objects = []
    for f in files:
        contents = open(f, "r").read()
        data = json_util.loads(contents)
        if isinstance(data, list):
            objects += data
        elif isinstance(data, dict):
            objects.append(data)
        else:
            logging.warning("JSON file {} did not contain an object or list of objects".format(f))
    return objects

项目：smappdragon 作者：SMAPPNYU | 项目源码 | 文件源码

def get_iterator(self):
        tweet_parser = TweetParser()
        if self.compression == 'bz2':
            self.mode = binary_mode(self.mode)
            json_handle = bz2.open(self.filepath, self.mode, encoding=self.encoding)
        elif self.compression == 'gzip':
            self.mode = binary_mode(self.mode)
            json_handle = gzip.open(self.filepath, self.mode, encoding=self.encoding)
        else:    
            json_handle = open(self.filepath, self.mode, encoding=self.encoding)
        bad_lines = 0
        for count, tweet in enumerate(json_handle):
            if not self.throw_error:
                try:
                    tweet = json_util.loads(tweet)
                except:
                    bad_lines += 1
            else:
                tweet = json_util.loads(tweet)
            if self.limit != 0 and self.limit <= count:
                return
            elif tweet_parser.tweet_passes_filter(self.filter, tweet) \
            and tweet_parser.tweet_passes_custom_filter_list(self.custom_filters, tweet):
                if self.should_strip:
                    yield tweet_parser.strip_tweet(self.keep_fields, tweet)
                else:
                    yield tweet
        if self.verbose:
            print("{} rows are ok.".format(count - bad_lines))
            print("{} rows are corrupt.".format(bad_lines))
        json_handle.close()

项目：webtzite 作者：materialsproject | 项目源码 | 文件源码

def _make_request(self, sub_url, payload=None, method="GET"):
        response = None
        url = self.preamble.replace('8000', '5000') + sub_url
        try:
            headers = {'Referer': self.preamble}
            if self.session.cookies.get('csrftoken') is None:
                from django.core.urlresolvers import reverse
                uri = urlparse.urlparse(self.preamble)
                domain = '{uri.scheme}://{uri.netloc}'.format(uri=uri).replace('8000', '5000')
                site_url = '/'.join(uri.path.split('/')[:-2]) # test_site/
                browserid_csrf = reverse('browserid.csrf')
                if site_url[:-1] not in browserid_csrf:
                    domain += site_url
                domain += browserid_csrf
                self.session.get(domain)
            headers["X-CSRFToken"] = self.session.cookies.get('csrftoken')
            response = self.session.request(method, url=url, headers=headers, data=payload)
            if response.status_code in [200, 400]:
                data = loads(response.text, json_options=JSONOptions(document_class=OrderedDict))
                if data["valid_response"]:
                    if data.get("warning"):
                        warnings.warn(data["warning"])
                    return data["response"]
                else:
                    raise MPResterError(data["error"])
            raise MPResterError(
                "REST query returned with error status code {}"
                .format(response.status_code)
            )
        except Exception as ex:
            msg = "{}. Content: {}".format(str(ex), repr(response.content)) \
                    if hasattr(response, "content") else str(ex)
            raise MPResterError(msg)

项目：picoCTF 作者：royragsdale | 项目源码 | 文件源码

def get_json_objects(files):
    objects = []
    for f in files:
        contents = open(f, "r").read()
        data = json_util.loads(contents)
        if isinstance(data, list):
            objects += data
        elif isinstance(data, dict):
            objects.append(data)
        else:
            logging.warning("JSON file {} did not contain an object or list of objects".format(f))
    return objects

项目：xgovctf 作者：alphagov | 项目源码 | 文件源码

def get_json_objects(files):
    objects = []
    for f in files:
        contents = open(f, "r").read()
        data = json_util.loads(contents)
        if isinstance(data, list):
            objects += data
        elif isinstance(data, dict):
            objects.append(data)
        else:
            logging.warning("JSON file {} did not contain an object or list of objects".format(f))
    return objects

项目：xgovctf 作者：alphagov | 项目源码 | 文件源码

def read_metadata(pid, n):
    """
    Reads the metadata object for a given problem instance.

    Args:
        pid: the problem id
        n: the problem instance
    Returns:
        The metadata object
    """

    metadata_path = get_metadata_path(pid, n)
    with open(metadata_path, "r") as f:
        return json_util.loads(f.read())