python类loads()的实例源码

def search_goods(limit_price, key_word):
    '''
    ???????????????????
    '''
    result = json_util.dumps(DB.Amazon.find(
        {'key_word': key_word,
         'prices.price': {'$gte': limit_price}},
        {'_id': 0}))
    return json_util.loads(result)

def find_porxy():
    # ?????????('state'==False), ???????????
    DB.ProxyIP.remove({'state': False})
    result = json_util.dumps(DB.ProxyIP.find({'state': True}, {'_id': 0}))
    return json_util.loads(result)

def auth_get_username(auth, user):
    """
    Get the username from a token
    """
    s = TimedJWSSerializer(app.config['SECRET_KEY'])

    token_user = s.loads(auth.get('password')).get('username')
    return token_user

def similar_funcs(self, text, typ=None):
        resp = requests.get(self.path + "/similar_funcs", json={"text":text, "type":typ})
        # resp is a list of snippet json data
        return bson_loads(resp.text)

def id2func(self, id_):
        resp = requests.get(self.path + "/id_to_func", params={"id":id_})
        # resp is a json snippet
        return bson_loads(resp.text)

def get_ids(self, ids):
        resp = requests.get(self.path + "/get_ids", json={"ids":ids})
        return bson_loads(resp.text)

def id2execution(self, id_):
        resp = requests.get(self.path + "/id_to_execution", params={"id":id_})
        # resp is a list of execution json
        return bson_loads(resp.text)

def id2bugs(self, id_):
        resp = requests.get(self.path + "/id_to_bugs", params={"id":id_})
        return bson_loads(resp.text)

def get_all_snippet_ids(self):
        resp = requests.get(self.path + "/all_snippet_ids")
        return bson_loads(resp.text)

def get_all_docstrings(self):
        resp = requests.get(self.path + "/all_docstrings")
        return bson_loads(resp.text)

def load(abspath, default=None, enable_verbose=True):
    """Load Json from file. If file are not exists, returns ``default``.

    :param abspath: file path. use absolute path as much as you can. 
      extension has to be ``.json`` or ``.gz`` (for compressed Json). 
    :type abspath: string

    :param default: default ``dict()``, if ``abspath`` not exists, return the
        default Python object instead.

    :param enable_verbose: default ``True``, help-message-display trigger.
    :type enable_verbose: boolean

    Usage::

        >>> from dataIO import js
        >>> js.load("test.json") # if you have a json file
        Load from 'test.json' ...
            Complete! Elapse 0.000432 sec.
        {'a': 1, 'b': 2}

    **????**

    ?Json???????

    :param abspath: Json??????, ????? ``.json`` ? ``.gz``, ?? ``.gz``
      ??????Json??
    :type abspath: ``???``

    :param default: ?? ``dict()``, ?????????, ??????????

    :param enable_verbose: ?? ``True``, ???????, ????????
    :type enable_verbose: ``???``
    """
    if default is None:
        default = dict()

    prt("\nLoad from '%s' ..." % abspath, enable_verbose)

    abspath = lower_ext(str(abspath))
    is_json = is_json_file(abspath)

    if not os.path.exists(abspath):
        prt("    File not found, use default value: %r" %
            default, enable_verbose)
        return default

    st = time.clock()
    if is_json:
        data = json.loads(textfile.read(abspath, encoding="utf-8"))
    else:
        data = json.loads(compress.read_gzip(abspath).decode("utf-8"))

    prt("    Complete! Elapse %.6f sec." % (time.clock() - st), enable_verbose)
    return data

def mq_callback(ch, method, properties, body) :

    body = loads(body)
    args = body.get("args")
    host = body.get("host")
    port = body.get("host")
    time = body.get("time")
    uuid = body.get("uuid")
    raw = body.get("raw")
    event = body.get("event")
    col.log.insert(body)
        result = None
        pprint(body)

        for rule in col.rules.find() :
            title = rule.get("title")
            condition = rule.get("condition")
            actions = rule.get("actions")
            execute = rule.get('exec')

            for k,v in condition.iteritems() :

                try :
                    test = args[int(k)] == v
                except :
                    pass

                if test :

                    print "Executing %s" % title
                    # See if there's an exec block
                    if execute :
                        typeof = execute.get('type')
                        valof = execute.get('value')
                        print "%s:%s" % (typeof, valof)

                        if typeof == 'JSON' :
                            try :
                                result = requests.post(valof, json=dumps(body)).json().get('result')
                                args.append(result)
                            except :
                                pass

                    for action in actions :
                        # this stupid thing just replaces vars in the string
                        argdict = dict(k for k in enumerate(args))
                        argdict.update({'result': result})
                        final = dict([(str(k), v) for k, v in argdict.items()])
                        try :
                            msg = action % final # {str(k) : v for k, v in enumerate(args)}
                        except :
                            return
                        print "MESSAGE : %s" % msg
                        for ms in msg.split("\n") :
                            sock.send(ms+"\r\n")