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

def execQuery (tree, query, aggcode=None):
    hosts = check_source (tree, query['name'])
    if send_source (hosts, tree, query['name']) == False:
        return []

    if aggcode:
        hosts = check_source (tree, aggcode['name'])
        if send_source (hosts, tree, aggcode['name']) == False:
            return []

    req = buildReq ('execQuery', tree, query, aggcode)
    resp, content = r.get (controller, json.dumps (req, default=json_util.default), "pathdump")
    if resp['status'] != '200':
        return []
    else:
        return json.loads (content, object_hook=json_util.object_hook)

def registerQuery (filepath):
    filename = os.path.basename (filepath)

    try:
        with open (filepath, 'r') as f:
            filedata = f.read()
    except EnvironmentError:
        return [False]

    req = {'api': 'registerQuery'}
    req.update ({'name': filename})
    req.update ({'data': filedata})

    resp, content = r.get (controller, json.dumps (req, default=json_util.default), "pathdump")
    if resp['status'] != '200':
        return []
    else:
        return json.loads (content, object_hook=json_util.object_hook)

def execRequest (req, url):
    global results
    workers = []
    tree = req['tree']
    for child in tree['controller']['child']:
        t = Thread (target = wrapper, args = (httpcmd, (child, req, url),
                                              results))
        workers.append (t)

    for worker in workers:
        worker.start()
    for worker in workers:
        worker.join()

    data = []
    for res in results:
        resp, content = res
        if resp['status'] == '200':
            data += json.loads (content, object_hook=json_util.object_hook)

    results = []
    return json.dumps (data, default=json_util.default)

def deserialize(value):
        return loads(value, object_hook=json_util.object_hook)

def installQuery (tree, query, interval):
    hosts = check_source (tree, query['name'])
    if send_source (hosts, tree, query['name']) == False:
        return []

    req = buildReq ('installQuery', tree, query, None, interval)
    resp, content = r.get (controller, json.dumps (req, default=json_util.default), "pathdump")
    if resp['status'] != '200':
        return []
    else:
        return json.loads (content, object_hook=json_util.object_hook)

def uninstallQuery (tree, query):
    req = buildReq ('uninstallQuery', tree, query)
    resp, content = r.get (controller, json.dumps (req, default=json_util.default), "pathdump")
    if resp['status'] != '200':
        return []
    else:
        return json.loads (content, object_hook=json_util.object_hook)

def check_source (tree, filename):
    req = {'api': 'check_source'}
    req.update ({'tree': tree})
    req.update ({'name': filename})

    resp, content = r.post (controller, json.dumps (req, default=json_util.default), "pathdump")
    return json.loads (content, object_hook=json_util.object_hook)

def getAggTree (groupnodes):
    req = {'api': 'getAggTree'}
    req.update ({'groupnodes': groupnodes})

    resp, content = r.get (controller, json.dumps (req, default=json_util.default), "pathdump")
    if resp['status'] != '200':
        return {}
    else:
        return json.loads (content, object_hook=json_util.object_hook)[0]

def getFlowCollectionDir():
    req = {'api': 'getFlowCollDir'}

    resp, content = r.get (controller, json.dumps (req, default=json_util.default), "pathdump")
    if resp['status'] != '200':
        return ''
    else:
        return json.loads (content, object_hook=json_util.object_hook)[0]

def run (argv, coll):
    flowid = argv[0]
    timeRange = json.loads (json.dumps(argv[1]),
                            object_hook=json_util.object_hook)
    if isinstance (timeRange[0], datetime.datetime):
        timeRange[0].replace (tzinfo=None)
    if isinstance (timeRange[1], datetime.datetime):
        timeRange[1].replace (tzinfo=None)
    linkID = ('*', '*')

    return pdapi.getPaths (flowid, linkID, timeRange)

def load(data):
    return json.loads(data, object_hook=object_hook)


# We don't want to have tzinfo in decoded json.This object_hook is
# overwritten json_util.object_hook for $date

def object_hook(dct):
    if "$date" in dct:
        dt = bson.json_util.object_hook(dct)
        return dt.replace(tzinfo=None)
    return bson.json_util.object_hook(dct)

def load_from_json(file):
    """
    This function load a json database into a list of dict.

    Parameters
    ----------
    file : file handler
        An already opened json file handler to the serialized job list.

    Returns
    -------
    list

    """
    job_list = list()
    city_dict = collections.defaultdict(int)
    for l in file.readlines():
        # use dict instead of directly object, better with pandas
        job = JobOfferAnon.from_json(json.loads(l, object_hook=json_util.object_hook)).to_dict()
        job['city'] = sanitize_city_name(job['city'])
        job['city'] = sanitize_city_name_for_geoloc(job['city'])
        city_file = pkg_resources.resource_filename('sfbistats.utils', 'city_locations.csv')
        dep, reg = city_to_dep_region(job['city'], city_file)
        job['department'] = dep
        job['region'] = reg
        job['duration'] = sanitize_duration(job['duration'])
        city_dict[job['city']] += 1
        job_list.append(job)
    job_list = spell_correct(job_list, city_dict)
    return job_list

def mongo_aggregate(cls, query, pipeline, hide_deleted=True):
        """Perform mongo aggregate queries
        query - is a dict which is to be passed to $match, a pipeline operator
        pipeline - list of dicts or dict of mongodb pipeline operators,
        http://docs.mongodb.org/manual/reference/operator/aggregation-pipeline
        """
        if isinstance(query, basestring):
            query = json.loads(
                query, object_hook=json_util.object_hook) if query else {}
        if not (isinstance(pipeline, dict) or isinstance(pipeline, list)):
            raise Exception(_(u"Invalid pipeline! %s" % pipeline))
        if not isinstance(query, dict):
            raise Exception(_(u"Invalid query! %s" % query))
        query = dict_for_mongo(query)
        if hide_deleted:
            # display only active elements
            deleted_at_query = {
                "$or": [{"_deleted_at": {"$exists": False}},
                        {"_deleted_at": None}]}
            # join existing query with deleted_at_query on an $and
            query = {"$and": [query, deleted_at_query]}
        k = [{'$match': query}]
        if isinstance(pipeline, list):
            k.extend(pipeline)
        else:
            k.append(pipeline)
        results = xform_instances.aggregate(k)
        return results['result']

def query_mongo_minimal(
            cls, query, fields, sort, start=0, limit=DEFAULT_LIMIT,
            count=False, hide_deleted=True):
        fields_to_select = {cls.USERFORM_ID: 0}
        # TODO: give more detailed error messages to 3rd parties
        # using the API when json.loads fails
        query = json.loads(
            query, object_hook=json_util.object_hook) if query else {}
        query = dict_for_mongo(query)
        if hide_deleted:
            # display only active elements
            # join existing query with deleted_at_query on an $and
            query = {"$and": [query, {"_deleted_at": None}]}
        # fields must be a string array i.e. '["name", "age"]'
        fields = json.loads(
            fields, object_hook=json_util.object_hook) if fields else []
        # TODO: current mongo (2.0.4 of this writing)
        # cant mix including and excluding fields in a single query
        if type(fields) == list and len(fields) > 0:
            fields_to_select = dict(
                [(_encode_for_mongo(field), 1) for field in fields])
        sort = json.loads(
            sort, object_hook=json_util.object_hook) if sort else {}
        cursor = xform_instances.find(query, fields_to_select)
        if count:
            return [{"count": cursor.count()}]

        if start < 0 or limit < 0:
            raise ValueError(_("Invalid start/limit params"))

        cursor.skip(start).limit(limit)
        if type(sort) == dict and len(sort) == 1:
            sort_key = sort.keys()[0]
            # TODO: encode sort key if it has dots
            sort_dir = int(sort[sort_key])  # -1 for desc, 1 for asc
            cursor.sort(_encode_for_mongo(sort_key), sort_dir)
        # set batch size
        cursor.batch_size = cls.DEFAULT_BATCHSIZE
        return cursor

def query_mongo(username, id_string, query=None, hide_deleted=True):
    print("incoming query", query)
    qry = query
    # query = None
    # query = json.loads(query, object_hook=json_util.object_hook)\
        # if query else {}
    # query = dict_for_mongo(query)
    # query[USERFORM_ID] = u'{0}_{1}'.format(username, id_string)
    # if hide_deleted:
        # query = {"$and": [query, {"_deleted_at": None}]}
    # query = {"$and": [query, qry]}
    print(qry)
    print("cpount", xform_instances.find(qry).count())
    return xform_instances.find(qry)

def _loads(content, fmt=None):
    if fmt == 'toml':
        return toml.loads(content)
    elif fmt == 'json':
        return json.loads(content, object_hook=json_util.object_hook)
    elif fmt == 'python':
        return ast.literal_eval(content)
    elif fmt == 'pickle':
        return pickle.loads(content)
    else:
        return content

def mongo_aggregate(cls, query, pipeline, hide_deleted=True):
        """Perform mongo aggregate queries
        query - is a dict which is to be passed to $match, a pipeline operator
        pipeline - list of dicts or dict of mongodb pipeline operators,
        http://docs.mongodb.org/manual/reference/operator/aggregation-pipeline
        """
        if isinstance(query, basestring):
            query = json.loads(
                query, object_hook=json_util.object_hook) if query else {}
        if not (isinstance(pipeline, dict) or isinstance(pipeline, list)):
            raise Exception(_(u"Invalid pipeline! %s" % pipeline))
        if not isinstance(query, dict):
            raise Exception(_(u"Invalid query! %s" % query))
        query = dict_for_mongo(query)
        if hide_deleted:
            # display only active elements
            deleted_at_query = {
                "$or": [{"_deleted_at": {"$exists": False}},
                        {"_deleted_at": None}]}
            # join existing query with deleted_at_query on an $and
            query = {"$and": [query, deleted_at_query]}
        k = [{'$match': query}]
        if isinstance(pipeline, list):
            k.extend(pipeline)
        else:
            k.append(pipeline)
        results = xform_instances.aggregate(k)
        return results['result']

def query_mongo_minimal(
            cls, query, fields, sort, start=0, limit=DEFAULT_LIMIT,
            count=False, hide_deleted=True):
        fields_to_select = {cls.USERFORM_ID: 0}
        # TODO: give more detailed error messages to 3rd parties
        # using the API when json.loads fails
        query = json.loads(
            query, object_hook=json_util.object_hook) if query else {}
        query = dict_for_mongo(query)
        if hide_deleted:
            # display only active elements
            # join existing query with deleted_at_query on an $and
            query = {"$and": [query, {"_deleted_at": None}]}
        # fields must be a string array i.e. '["name", "age"]'
        fields = json.loads(
            fields, object_hook=json_util.object_hook) if fields else []
        # TODO: current mongo (2.0.4 of this writing)
        # cant mix including and excluding fields in a single query
        if type(fields) == list and len(fields) > 0:
            fields_to_select = dict(
                [(_encode_for_mongo(field), 1) for field in fields])
        sort = json.loads(
            sort, object_hook=json_util.object_hook) if sort else {}
        cursor = xform_instances.find(query, fields_to_select)
        if count:
            return [{"count": cursor.count()}]

        if start < 0 or limit < 0:
            raise ValueError(_("Invalid start/limit params"))

        cursor.skip(start).limit(limit)
        if type(sort) == dict and len(sort) == 1:
            sort_key = sort.keys()[0]
            # TODO: encode sort key if it has dots
            sort_dir = int(sort[sort_key])  # -1 for desc, 1 for asc
            cursor.sort(_encode_for_mongo(sort_key), sort_dir)
        # set batch size
        cursor.batch_size = cls.DEFAULT_BATCHSIZE
        return cursor

def query_mongo(username, id_string, query=None, hide_deleted=True):
    query = json.loads(query, object_hook=json_util.object_hook)\
        if query else {}
    query = dict_for_mongo(query)
    query[USERFORM_ID] = u'{0}_{1}'.format(username, id_string)
    if hide_deleted:
        # display only active elements
        # join existing query with deleted_at_query on an $and
        query = {"$and": [query, {"_deleted_at": None}]}
    return xform_instances.find(query)

def handleRequest (req):
    global query_results

    Tree = req['tree']
    cur = helper.getCurNodeID ()
    if len (Tree[cur]['child']) == 0:
        return helper.handleLeafNode (req)

    # From now on, the following handles when the current node is a relay node
    workers = []
    # 1) create a worker thread at the current node
    (func, argv) = helper.getThreadArgument (True, req)
    t = Thread (target = helper.wrapper, args = (func, argv, query_results))
    workers.append (t)

    # 2) deliver query to child nodes
    for child in Tree[cur]['child']: 
        (func, argv) = helper.getThreadArgument (False, req, child)
        # further optimization (should be implemented): construct a subtree for
        # each child and pass it on to the httpcmd as argument
        t = Thread (target = helper.wrapper, args = (func, argv,
                                                     query_results))
        workers.append (t)

    # 3) start workers
    for worker in workers:
        worker.start()

    # 4) wait unitl workers finish -> this part might be hung forever
    for worker in workers:
        worker.join()

    data=[]
    for res in query_results:
        if len(res) > 0 and type(res) == type(()) and 'content-type' in res[0]:
            resp, content = res
            content = json.loads (content, object_hook=json_util.object_hook)
        else:
            content = res
        data += content
    # reset variables 
    query_results = []

    if req['api'] == 'execQuery' and 'aggcode' in req:
        # 4) process collected data using AggCode
        return helper.processCollectedData (req['aggcode'], data)
    else:
        return data

def query_mongo(cls, username, id_string, query, fields, sort, start=0,
                    limit=DEFAULT_LIMIT, count=False, hide_deleted=True):
        fields_to_select = {cls.USERFORM_ID: 0}
        # TODO: give more detailed error messages to 3rd parties
        # using the API when json.loads fails
        if isinstance(query, basestring):
            query = json.loads(query, object_hook=json_util.object_hook)
        query = query if query else {}
        query = dict_for_mongo(query)
        query[cls.USERFORM_ID] = u'%s_%s' % (username, id_string)

        # check if query contains and _id and if its a valid ObjectID
        if '_uuid' in query and ObjectId.is_valid(query['_uuid']):
            query['_uuid'] = ObjectId(query['_uuid'])

        if hide_deleted:
            # display only active elements
            # join existing query with deleted_at_query on an $and
            query = {"$and": [query, {"_deleted_at": None}]}

        # fields must be a string array i.e. '["name", "age"]'
        if isinstance(fields, basestring):
            fields = json.loads(fields, object_hook=json_util.object_hook)
        fields = fields if fields else []

        # TODO: current mongo (2.0.4 of this writing)
        # cant mix including and excluding fields in a single query
        if type(fields) == list and len(fields) > 0:
            fields_to_select = dict(
                [(_encode_for_mongo(field), 1) for field in fields])
        if isinstance(sort, basestring):
            sort = json.loads(sort, object_hook=json_util.object_hook)
        sort = sort if sort else {}

        cursor = xform_instances.find(query, fields_to_select)
        if count:
            return [{"count": cursor.count()}]

        if start < 0 or limit < 0:
            raise ValueError(_("Invalid start/limit params"))

        cursor.skip(start).limit(limit)
        if type(sort) == dict and len(sort) == 1:
            sort_key = sort.keys()[0]
            # TODO: encode sort key if it has dots
            sort_dir = int(sort[sort_key])  # -1 for desc, 1 for asc
            cursor.sort(_encode_for_mongo(sort_key), sort_dir)
        # set batch size
        cursor.batch_size = cls.DEFAULT_BATCHSIZE
        return cursor

def query_mongo(cls, username, id_string, query, fields, sort, start=0,
                    limit=DEFAULT_LIMIT, count=False, hide_deleted=True):
        fields_to_select = {cls.USERFORM_ID: 0}
        # TODO: give more detailed error messages to 3rd parties
        # using the API when json.loads fails
        if isinstance(query, basestring):
            query = json.loads(query, object_hook=json_util.object_hook)
        query = query if query else {}
        query = dict_for_mongo(query)
        query[cls.USERFORM_ID] = u'%s_%s' % (username, id_string)

        # check if query contains and _id and if its a valid ObjectID
        if '_uuid' in query and ObjectId.is_valid(query['_uuid']):
            query['_uuid'] = ObjectId(query['_uuid'])

        if hide_deleted:
            # display only active elements
            # join existing query with deleted_at_query on an $and
            query = {"$and": [query, {"_deleted_at": None}]}

        # fields must be a string array i.e. '["name", "age"]'
        if isinstance(fields, basestring):
            fields = json.loads(fields, object_hook=json_util.object_hook)
        fields = fields if fields else []

        # TODO: current mongo (2.0.4 of this writing)
        # cant mix including and excluding fields in a single query
        if type(fields) == list and len(fields) > 0:
            fields_to_select = dict(
                [(_encode_for_mongo(field), 1) for field in fields])
        if isinstance(sort, basestring):
            sort = json.loads(sort, object_hook=json_util.object_hook)
        sort = sort if sort else {}

        cursor = xform_instances.find(query, fields_to_select)
        if count:
            return [{"count": cursor.count()}]

        if start < 0 or limit < 0:
            raise ValueError(_("Invalid start/limit params"))

        cursor.skip(start).limit(limit)
        if type(sort) == dict and len(sort) == 1:
            sort_key = sort.keys()[0]
            # TODO: encode sort key if it has dots
            sort_dir = int(sort[sort_key])  # -1 for desc, 1 for asc
            cursor.sort(_encode_for_mongo(sort_key), sort_dir)
        # set batch size
        cursor.batch_size = cls.DEFAULT_BATCHSIZE
        return cursor