Python ujson 模块，loads() 实例源码

def query(self, hostname, type='A', dnssec=True):
        assert utils.validate_hostname(hostname)
        assert utils.validate_rr_type(type)

        if (type in ('PTR', RR['PTR']) and
            not (hostname.endswith('.in-addr.arpa') or
                 hostname.endswith('.in-addr.arpa.'))):
            hostname = '%s.in-addr.arpa' % (hostname)

        params = self.build_params(hostname, type, dnssec)
        url = '%s?%s' % (self.server, params)
        req = self.session.request('GET', url,
                                   headers=PublicDNS.default_headers)
        resp = self.session.get_response(req)
        if resp.status != 200:
            raise InvalidHTTPStatusCode
        body = resp.read()
        json = load_json(body)
        obj = utils.populate_response(json)
        return obj

def load_body(self):
        """ Load http request body and returns
            form data and files.
        """
        environ = self.environ
        cl = environ['CONTENT_LENGTH']
        icl = int(cl)
        if icl > self.max_content_lenght:
            raise ValueError('Maximum content length exceeded')
        fp = environ['wsgi.input']
        ct = environ['CONTENT_TYPE']
        # application/x-www-form-urlencoded
        if '/x' in ct:
            return parse_qs(fp.read(icl).decode(self.encoding)), None
        # application/json
        elif '/j' in ct:
            return json_loads(fp.read(icl).decode(self.encoding)), None
        # multipart/form-data
        elif ct.startswith('m'):
            return parse_multipart(fp, ct, cl, self.encoding)
        else:
            return None, None

def do_POST(self):
        self.logger.debug('Webhook triggered')
        try:
            self._validate_post()
            clen = self._get_content_len()
        except _InvalidPost as e:
            self.send_error(e.http_code)
            self.end_headers()
        else:
            buf = self.rfile.read(clen)
            json_string = bytes_to_native_str(buf)

            self.send_response(200)
            self.end_headers()

            self.logger.debug('Webhook received data: ' + json_string)

            update = Update.de_json(json.loads(json_string), self.server.bot)

            self.logger.debug('Received Update with ID %d on Webhook' % update.update_id)
            self.server.update_queue.put(update)

def get_queryset(self, req, resp):
        query_term = self.get_param_or_post(req, self.PARAM_TEXT_QUERY)
        search = self.get_param_or_post(req, self.PARAM_SEARCH)
        if search:
            try:
                req.params['__raw__'] = json.loads(search)
            except ValueError:
                raise HTTPBadRequest('Invalid attribute',
                                     'Value of {} filter attribute is invalid'.format(self.PARAM_SEARCH))
        order = self.get_param_or_post(req, self.PARAM_ORDER)
        queryset = self.objects_class.objects(**req.params)
        if query_term is not None:
            queryset = queryset.search_text(query_term)
        if order:
            queryset = queryset.order_by(order)
        return queryset

def get_param_totals(self, req):
        """
        Gets the totals and total_count params and normalizes them into a single list.

        :param req: Falcon request
        :type req: falcon.request.Request

        :return: total expressions
        :rtype: list
        """
        totals = self.get_param_or_post(req, self.PARAM_TOTALS, [])
        if totals:
            if isinstance(totals, str):
                totals = json.loads(totals)
            if isinstance(totals, dict):
                totals = [totals]
            else:
                totals = list(map(lambda x: x if isinstance(x, dict) else {x: None}, totals))
        total_count = self.get_param_or_post(req, self.PARAM_TOTAL_COUNT)
        if total_count and not list(filter(lambda x: 'count' in x, totals)):
            totals.append({'count': None})
        return totals

def test_integrity_error_handling_with_foreign_key(self, post_method, stream, session):
        stream.feed_data(ujson.dumps([{'m2_id': 1}]).encode())
        stream.feed_eof()
        request = SwaggerRequest('/model1/', 'post', body=stream, headers={'content-type': 'application/json'})
        resp = await post_method(request, session)

        assert resp.status_code == 400
        assert ujson.loads(resp.body) == {
            'params': {'m2_id': 1},
            'database message': {
                'message':  'Cannot add or update a child row: '
                            'a foreign key constraint fails '
                            '(`swaggerit_test`.`model1_swagger`, '
                            'CONSTRAINT `model1_swagger_ibfk_1` FOREIGN '
                            'KEY (`m2_id`) REFERENCES `model2_swagger` '
                            '(`id`))',
                'code': 1452
            }
        }

def test_model_base_error_handling_with_patch_and_with_nested_delete(self, patch_method, post_method, stream, session):
        stream.feed_data(b'[{}]')
        stream.feed_eof()
        request = SwaggerRequest('/model1/1/', 'patch', body=stream, headers={'content-type': 'application/json'})
        await post_method(request, session)

        stream = asyncio.StreamReader(loop=session.loop)
        body = {'model2_': {'id': 1, '_operation': 'delete'}}
        stream.feed_data(ujson.dumps(body).encode())
        stream.feed_eof()
        request = SwaggerRequest('/model1/1/', 'patch', path_params={'id': 1}, body=stream, headers={'content-type': 'application/json'})
        resp = await patch_method(request, session)

        assert resp.status_code == 400
        assert ujson.loads(resp.body) == {
            'instance': [body],
            'message': "Can't execute nested 'delete' operation"
        }

def test_model_base_error_handling_with_patch_and_with_nested_remove(self, patch_method, post_method, stream, session):
        stream.feed_data(b'[{}]')
        stream.feed_eof()
        request = SwaggerRequest('/model1/1/', 'patch', body=stream, headers={'content-type': 'application/json'})
        await post_method(request, session)

        stream = asyncio.StreamReader(loop=session.loop)
        body = {'model2_': {'id': 1, '_operation': 'remove'}}
        stream.feed_data(ujson.dumps(body).encode())
        stream.feed_eof()
        request = SwaggerRequest('/model1/1/', 'patch', path_params={'id': 1}, body=stream, headers={'content-type': 'application/json'})
        resp = await patch_method(request, session)

        assert resp.status_code == 400
        assert ujson.loads(resp.body) == {
            'instance': [body],
            'message': "Can't execute nested 'remove' operation"
        }

def test_model_base_error_handling_with_patch_and_with_nested_update(self, patch_method, post_method, stream, session):
        stream.feed_data(b'[{}]')
        stream.feed_eof()
        request = SwaggerRequest('/model1/1/', 'patch', body=stream, headers={'content-type': 'application/json'})
        await post_method(request, session)

        stream = asyncio.StreamReader(loop=session.loop)
        body = {'model2_': {'id': 1, '_operation': 'update'}}
        stream.feed_data(ujson.dumps(body).encode())
        stream.feed_eof()
        request = SwaggerRequest('/model1/1/', 'patch', path_params={'id': 1}, body=stream, headers={'content-type': 'application/json'})
        resp = await patch_method(request, session)

        assert resp.status_code == 400
        assert ujson.loads(resp.body) == {
            'instance': [body],
            'message': "Can't execute nested 'update' operation"
        }

def _post(self, uri, data):
        if type(data) == dict:
            data = [data]

        if type(data[0]) != dict:
            raise RuntimeError('submitted data must be a dictionary')

        data = json.dumps(data)

        if self.nowait:
            uri = "{0}?nowait=1".format(uri)

        logger.debug('uri: %s' % uri)

        body = self.session.post(uri, data=data, verify=self.verify_ssl)
        logger.debug('status code: ' + str(body.status_code))
        if body.status_code > 299:
            logger.error('request failed: %s' % str(body.status_code))
            logger.error(json.loads(body.text).get('message'))
            return None

        body = json.loads(body.text)
        return body

def on_message(self, message):
        data = json.loads(message)

        if not isinstance(data, list):
            raise Exception('incoming data is no list')

        if data[0] == self.MSG_PREFIX and len(data) == 3:
            prefix, uri = data[1:3]
            self.prefixes.add(prefix, uri)

        elif data[0] == self.MSG_CALL and len(data) >= 3:
            return self.rpc_call(data)

        elif data[0] in (self.MSG_SUBSCRIBE, self.MSG_UNSUBSCRIBE,
                         self.MSG_PUBLISH):
            return self.pubsub_action(data)
        else:
            raise Exception("Unknown call")

def consume_messages(cls, **kwargs):
        def job(consumer_, redis_client_, redis_channel_):
            for msg in consumer_.poll():
                message = msg.value
                logger.info(ujson.loads(message))
                redis_client_.publish(redis_channel_, message)

        def shut_down(consumer_):
            consumer_.shut_down()

        # get consumer
        kafka_broker = kwargs.get(KAFKA_BROKER) or DEFAULT_KAFKA_BROKER
        kafka_topic = kwargs.get(KAFKA_OUTPUT_TOPIC) or DEFAULT_KAFKA_OUTPUT_TOPIC
        consumer = Consumer(kafka_broker, kafka_topic)

        # get redis
        redis_channel = kwargs.get(REDIS_CHANNEL) or DEFAULT_REDIS_CHANNEL
        redis_host = kwargs.get(REDIS_HOST) or DEFAULT_REDIS_HOST
        redis_port = kwargs.get(REDIS_PORT) or DEFAULT_REDIS_PORT
        redis_client = redis.StrictRedis(host=redis_host, port=redis_port)

        atexit.register(shut_down, consumer)

        scheduler = Scheduler(1, job, consumer, redis_client, redis_channel)
        scheduler.run()

def test_storage():
    app = Sanic('test_text')

    @app.middleware('request')
    def store(request):
        request['user'] = 'sanic'
        request['sidekick'] = 'tails'
        del request['sidekick']

    @app.route('/')
    def handler(request):
        return json({ 'user': request.get('user'), 'sidekick': request.get('sidekick') })

    request, response = sanic_endpoint_test(app)

    response_json = loads(response.text)
    assert response_json['user'] == 'sanic'
    assert response_json.get('sidekick') is None

def process_request(self, req, resp):
        # req.stream corresponds to the WSGI wsgi.input environ variable,
        # and allows you to read bytes from the request body.
        #
        # See also: PEP 3333
        if req.content_length in (None, 0):
            # Nothing to do
            return

        body = req.stream.read()
        if not body:
            raise falcon.HTTPBadRequest('Empty request body',
                                        'A valid JSON document is required.')

        try:
            req.context['doc'] = json.loads(body.decode('utf-8'))

        except (ValueError, UnicodeDecodeError):
            raise falcon.HTTPError(falcon.HTTP_753,
                                   'Malformed JSON',
                                   'Could not decode the request body. The '
                                   'JSON was incorrect or not encoded as '
                                   'UTF-8.')

def handleSetOff(self,request):
        path = request.path
        value = json.loads(request.body)
        group     = value["group"]
        onoff     = value["onoff"]
        log.debug ("Group: %s Set motors %s",group,onoff)
        if "L" in group:
            driver = driverL
        if "R" in group:
            driver = driverR

        if onoff == '-1':
            driver.allExit()
        elif onoff == '0':    
            driver.allOff()
        elif onoff == '1':    
            driver.allOn()
        yield from request.sendOk()

def _find_layers(img, id):
    with closing(img.extractfile('%s/json' % id)) as fd:
        f_content = fd.read()
        if hasattr(f_content, "decode"):
            f_content = f_content.decode()
        info = json.loads(f_content)

    log.debug('layer = %s', id)
    for k in ['os', 'architecture', 'author', 'created']:
        if k in info:
            log.debug('%s = %s', k, info[k])

    yield id

    if 'parent' in info:
        pid = info['parent']
        for layer in _find_layers(img, pid):
            yield layer


# --------------------------------------------------------------------------
# Public API
# --------------------------------------------------------------------------

def get_root_json_from_image(img: tarfile.TarFile) -> Tuple[str, dict]:
    """
    Every docker image has a root .json file with the metadata information.
    this function locate this file, load it and return the value of it and
    their name

    >>> get_docker_image_layers(img)
    ('db079554b4d2f7c65c4df3adae88cb72d051c8c3b8613eb44e86f60c945b1ca7', dict(...))
    """
    for f in img.getmembers():
        if f.name.endswith("json") and "/" not in f.name:
            c = img.extractfile(f.name).read()
            if hasattr(c, "decode"):
                c = c.decode()

            return f.name.split(".")[0], json.loads(c)

    return None, None

def onMessage(self, payload, isBinary):
        extra_logger.debug('[%s] Received payload: %s', self.factory.label, payload)
        assert not isBinary
        payload = ujson.loads(payload)

        context = self._make_context()
        latency = context['start'] - payload['headers']['sent_at']
        pyprofile.incr('rewarder_protocol.messages')
        pyprofile.incr('rewarder_protocol.messages.{}'.format(payload['method']))

        # Double latency to model RTT
        pyprofile.timing('rewarder_protocol.latency.rtt.skew_unadjusted', 2*latency)
        if latency < 0:
            pyprofile.incr('rewarder_protocol.latency.rtt.skew_unadjusted.negative')

        self.recv(context, payload)

def test_double_performance_float_precision(benchmark):
    print("\nArray with 256 doubles:")
    name = 'rapidjson (precise)'
    serialize = rapidjson.dumps
    deserialize = rapidjson.loads

    ser_data, des_data = benchmark(run_client_test,
                                   name, serialize, deserialize,
                                   data=doubles,
                                   iterations=50000,
                                   )
    msg = "%-11s serialize: %0.3f  deserialize: %0.3f  total: %0.3f" % (
        name, ser_data, des_data, ser_data + des_data
    )

    print(msg)

def from_tweets(cls, tweets, metadata=None, **kwargs):
        """
        :param tweets: a iterable of tweets
        :param kwargs: extra attributes to be considered for inclusion. should be json serializable.
        :return:
        """

        tl = cls()

        json_tweets = json.loads(serializers.serialize("json", tweets));
        for key, values in kwargs.items():
            if len(values) != len(json_tweets):
                continue

            for tweet, value in zip(json_tweets, values):
                tweet['fields'][key] = value

        tl.save()

        json_repr = {'metadata': metadata, 'tweets': json_tweets, 'pk': tl.pk, 'created_at': tl.datetime.isoformat()}
        tl.json = json.dumps(json_repr)
        tl.save()
        return tl

def test_should_report_healthy_if_kafka_healthy(self, kafka_check):
        kafka_check.healthcheck.return_value = healthcheck.CheckResult(True,
                                                                       'OK')
        self.resource._kafka_check = kafka_check

        ret = self.simulate_request(ENDPOINT,
                                    headers={
                                        'Content-Type': 'application/json'
                                    },
                                    decode='utf8',
                                    method='GET')
        self.assertEqual(falcon.HTTP_OK, self.srmock.status)

        ret = json.loads(ret)
        self.assertIn('kafka', ret)
        self.assertEqual('OK', ret.get('kafka'))

def test_should_report_unhealthy_if_kafka_unhealthy(self, kafka_check):
        url = 'localhost:8200'
        err_str = 'Could not connect to kafka at %s' % url
        kafka_check.healthcheck.return_value = healthcheck.CheckResult(False,
                                                                       err_str)
        self.resource._kafka_check = kafka_check

        ret = self.simulate_request(ENDPOINT,
                                    headers={
                                        'Content-Type': 'application/json'
                                    },
                                    decode='utf8',
                                    method='GET')
        self.assertEqual(falcon.HTTP_SERVICE_UNAVAILABLE, self.srmock.status)

        ret = json.loads(ret)
        self.assertIn('kafka', ret)
        self.assertEqual(err_str, ret.get('kafka'))

def genericLazyConfig(db, dbKey, serviceName, clusterName):
    if db.exists(dbKey):
        rawConfig = db.get(dbKey)

        try:
            config = json.loads(rawConfig)
        except ValueError as e:
            raise ValueError("Failed to decode SQL DB config %s: %s" % (config, e))

        try:
            host = config["host"]
            database = config["database"]
            user = config["user"]
            password = config["password"]
        except KeyError as e:
            raise KeyError("Missing a required SQL config key in config %s: %s" % (config, e))

        return host, database, user, password
    else:
        raise RuntimeError("No SQL connection configures for "
                           "service %s of cluster %s" % (
                               serviceName, clusterName))

def notifySpaceModification(self, id, **args):
        """
        param:id id of space which changed
        result bool 

        """
        id=id.lower()
        loaders = j.core.portal.active.spacesloader
        loader = loaders.getLoaderFromId(id)
        loader.reset()

        ctx=args["ctx"]

        if "payload" in ctx.params:

            payload=ujson.loads(ctx.params["payload"])

            owner=payload["repository"]["owner"]
            name=payload["repository"]["name"]

            cmd="cd /opt/code/%s/%s;hg pull;hg update -C"%(owner,name)
            print("execute %s"%cmd)
            j.system.process.execute(cmd)

def main(j, args, params, tags, tasklet):
    params.merge(args)
    doc = params.doc
    data = args.getTag('data')
    title = args.getTag('title')

    out = "*%s*\n" % title
    try:    
        objargs = json.loads(data)
        for key,value in objargs.iteritems():
            if not value:
                value = ''
            out += "|%s|%s|\n"%(str(key),j.html.escape(str(value)))
    except Exception:
        out = ''
    params.result = (out, doc)
    return params

def create_random_vote(self):
        """Create secrrets for voting."""
        """This is done for prevent automatic or robotic voting."""
        random_number = random.randrange(0, 100000)
        current_time = int(time.time())
        result = await self.db.fetchrow(
            # 'select * from stickers order by random() limit 1;'
            'SELECT * FROM stickers TABLESAMPLE SYSTEM_ROWS(1);'
        )
        random_sticker = json.loads(result[1])
        token = await self.db.fetchval(
            "select md5('{}');".format(random_number))
        await self.db.fetch(
            "insert into secret (data) values"
            "('{}')".format(json.dumps([
                token,
                current_time,
                random_sticker[0],
                random_sticker[2]
            ])))
        return (random_sticker[2], token)

def get_statistics(self):
        sql = "select data from server"
        server = await self.db.fetchrow(sql)
        server = json.loads(server[0])
        sql = "SELECT count (data) FROM stickers"
        stickers_count = await self.db.fetchval(sql)
        sql = "SELECT count (data) FROM packs"
        packs_count = await self.db.fetchval(sql)
        statistics = {
            "users": server[1],
            "clicks": server[2],
            "votes": server[3],
            "packs_count": packs_count,
            "stickers_count": stickers_count,
        }
        logging.debug('Get statistics: {}'.format(str(statistics)))
        return statistics

    # Bellow functions related to Stickers API and don't using by main site

def on_message(self, message):
        data = json.loads(message)

        if not isinstance(data, list):
            raise Exception('incoming data is no list')

        if data[0] == self.MSG_PREFIX and len(data) == 3:
            prefix, uri = data[1:3]
            self.prefixes.add(prefix, uri)

        elif data[0] == self.MSG_CALL and len(data) >= 3:
            return self.rpc_call(data)

        elif data[0] in (self.MSG_SUBSCRIBE, self.MSG_UNSUBSCRIBE,
                         self.MSG_PUBLISH):
            return self.pubsub_action(data)
        else:
            raise Exception("Unknown call")

def check_json(json_type):
        """
        Checks whether json_type is a dict or a string. If it is already a dict, it is returned as-is.
        If it is not, it is converted to a dict by means of json.loads(json_type)
        :param json_type:
        :return:
        """
        try:
            str_types = (str, unicode)
        except NameError:
            str_types = (str,)

        if type(json_type) == dict:
            return json_type
        elif type(json_type) in str_types:
            return json.loads(json_type)
        else:
            raise ValueError("json_type should be a json dict or string.")

def _request(method, path, type, **kwargs):
    url = 'http://localhost:8000%s' % path
    res = method(url, **kwargs)
    res.raise_for_status()
    if type == 'html':
        assert 'text/html' in res.headers['Content-Type']
        doc = bs4.BeautifulSoup(res.text, 'html5lib')
    elif type == 'text':
        assert 'text/plain' in res.headers['Content-Type']
        doc = res.text
    elif type == 'json':
        assert ('application/json' in res.headers['Content-Type'] or
                'text/plain' in res.headers['Content-Type'])
        doc = ujson.loads(res.text)
    else:
        assert False, type
    return (res, doc)

def query_jql(self, script, params=None):
        """Query the Mixpanel JQL API

        https://mixpanel.com/help/reference/jql/api-reference#api/access

        :param script: String containing a JQL script to run
        :param params: Optional dict that will be made available to the script as the params global variable.
        :type script: str
        :type params: dict

        """
        query_params = {"script": script}
        if params is not None:
            query_params["params"] = json.dumps(params)

        response = self.request(Mixpanel.FORMATTED_API, ['jql'], query_params, method='POST')
        return json.loads(response)

def _get_engage_page(self, params):
        """Fetches and returns the response from an /engage request

        :param params: Query parameters for the /engage API
        :type params: dict
        :return: /engage API response object
        :rtype: dict

        """
        response = self.request(Mixpanel.FORMATTED_API, ['engage'], params)
        data = json.loads(response)
        if 'results' in data:
            return data
        else:
            Mixpanel.LOGGER.warning("Invalid response from /engage: " + response)
            return

def calculate_query_bin_bits(tfidf): #this also needs to return the table from redis as well as the bin id
    table = str2int( ujson.loads( r.get('table') ) )
    dim = int( r.get('dim') )
    mapping = ujson.loads( r.get('map') )
    mapping = pd.DataFrame({'word': mapping})
    num_vectors = 16

    words = list(tfidf.keys())
    values = list(tfidf.values())
    tfidf_df = pd.DataFrame({'word': words, 'value': values})

    article_representation = pd.merge(mapping, tfidf_df, on='word', how='left').fillna(0)['value']

    bin_vectors = generate_random_vectors(num_vectors, dim)
    powers_of_two = 1 << np.arange(num_vectors-1, -1, -1)
    query_bin_bits = (article_representation.dot(bin_vectors) >= 0)

    return query_bin_bits, table

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 = ujson.dumps(data)

        self.assertEqual(data["int"], ujson.loads(js)["int"])
        self.assertAlmostEqual(data["float"], ujson.loads(js)[
                               "float"], delta=0.0001)
        self.assertEqual(data["str"], ujson.loads(js)["str"])
        self.assertNotEqual(data["bytes"], ujson.loads(js)["bytes"])  # ???
        self.assertEqual(data["boolean"], ujson.loads(js)["boolean"])
        self.assertNotEqual(data["datetime"], ujson.loads(js)["datetime"])

        print(ujson.dumps(data, indent=4))


#--- Unittest ---

def process_target_event(self, event, msg):
        """Process Target.* dev tools events"""
        if event == 'attachedToTarget':
            if 'targetInfo' in msg['params'] and 'targetId' in msg['params']['targetInfo']:
                target = msg['params']['targetInfo']
                if 'type' in target and target['type'] == 'service_worker':
                    self.workers.append(target)
                    if self.recording:
                        self.send_command('Network.enable', {}, target_id=target['targetId'])
                self.send_command('Runtime.runIfWaitingForDebugger', {},
                                  target_id=target['targetId'])
        if event == 'receivedMessageFromTarget':
            if 'message' in msg['params'] and 'targetId' in msg['params']:
                logging.debug(msg['params']['message'][:200])
                target_id = msg['params']['targetId']
                target_message = json.loads(msg['params']['message'])
                self.process_message(target_message, target_id=target_id)

def process_pcap(self):
        """Process the pcap in a background thread"""
        pcap_file = self.pcap_file + '.gz'
        if os.path.isfile(pcap_file):
            path_base = os.path.join(self.task['dir'], self.task['prefix'])
            slices_file = path_base + '_pcap_slices.json.gz'
            pcap_parser = os.path.join(os.path.abspath(os.path.dirname(__file__)),
                                       'support', "pcap-parser.py")
            cmd = ['python', pcap_parser, '--json', '-i', pcap_file, '-d', slices_file]
            logging.debug(cmd)
            try:
                stdout = subprocess.check_output(cmd)
                if stdout is not None:
                    result = json.loads(stdout)
                    if result:
                        if 'in' in result:
                            self.task['page_data']['pcapBytesIn'] = result['in']
                        if 'out' in result:
                            self.task['page_data']['pcapBytesOut'] = result['out']
                        if 'in_dup' in result:
                            self.task['page_data']['pcapBytesInDup'] = result['in_dup']
            except Exception:
                pass

def post(self):
        """Handle POST messages"""
        import ujson as json
        try:
            messages = self.request.body
            if messages is not None and len(messages):
                if self.request.uri == '/log':
                    logging.debug(messages)
                else:
                    for line in messages.splitlines():
                        line = line.strip()
                        if len(line):
                            message = json.loads(line)
                            if 'body' not in message and self.request.uri != '/etw':
                                message['body'] = None
                            MESSAGE_SERVER.handle_message(message)
        except Exception:
            pass
        self.set_status(200)

def _inner_iter_compactsplit(line, skip_callback):
    #print ('CSLINE',line[:40])
    #t0 = time()
    new_id, dd = line.strip('\n').split('\t', 1)

    rec = u_loads(dd)

    if skip_callback is not False:
        ## Return False to skip:
        sk = skip_callback(rec['_id'])
        if sk is False:
            ## todo re-add fastforward
            return False

        got_any = True

        #print ('INNER',(time() - t0) * 1000, 'ms')

        #yield sk, rec
        return sk, rec
    else:
        #print ('INNER',(time() - t0) * 1000, 'ms')
        #yield rec
        return rec

def Cache(redis_key, expire=15 * 60, use_cache=True):
    """??????
    """
    def deco(func):
        def wrapper(self, *args, **kwargs):
            if len(args) >= 1:
                REDIS_KEY = redis_key.format(*args)
            else:
                REDIS_KEY = redis_key
            redis_conn = get_redis("main")
            cache_str = redis_conn.get(REDIS_KEY)
            if cache_str and use_cache and config_cache:
                cache = ujson.loads(cache_str)
                return cache

            cache = func(self, *args, **kwargs)
            cache_str = ujson.dumps(cache)
            redis_conn.set(REDIS_KEY, cache_str, expire)

            return cache
        return wrapper
    return deco

def test_storage():
    app = Sanic('test_text')

    @app.middleware('request')
    def store(request):
        request['user'] = 'sanic'
        request['sidekick'] = 'tails'
        del request['sidekick']

    @app.route('/')
    def handler(request):
        return json({'user': request.get('user'), 'sidekick': request.get('sidekick')})

    request, response = app.test_client.get('/')

    response_json = loads(response.text)
    assert response_json['user'] == 'sanic'
    assert response_json.get('sidekick') is None

def test_app_injection():
    app = Sanic('test_app_injection')
    expected = random.choice(range(0, 100))

    @app.listener('after_server_start')
    async def inject_data(app, loop):
        app.injected = expected

    @app.get('/')
    async def handler(request):
        return json({'injected': request.app.injected})

    request, response = app.test_client.get('/')

    response_json = loads(response.text)
    assert response_json['injected'] == expected

def on_message(self, message):
        data = json.loads(message)

        if not isinstance(data, list):
            raise Exception('incoming data is no list')

        if data[0] == self.MSG_PREFIX and len(data) == 3:
            prefix, uri = data[1:3]
            self.prefixes.add(prefix, uri)

        elif data[0] == self.MSG_CALL and len(data) >= 3:
            return self.rpc_call(data)

        elif data[0] in (self.MSG_SUBSCRIBE, self.MSG_UNSUBSCRIBE,
                         self.MSG_PUBLISH):
            return self.pubsub_action(data)
        else:
            raise Exception("Unknown call")

def test_get_task_by_valid_uuid(self):
        filename = os.path.join(self.samples, "sample.docx")
        response = self.submit_file(filename, {
            "formats": ["txt"]
        })
        json = ujson.loads(response.data)
        self.assertEqual(response.status_code, 200)
        self.assertTrue(json.get("id"))
        self.assertEqual(json.get("status"), "queued")
        response = self.client.get("/api/v1/{0}".format(json.get("id")))
        self.assertEqual(response.status_code, 200)
        self.assertEqual(ujson.loads(response.data), {
            "id": json.get("id"),
            "status": "queued",
            "result_url": None,
        })

def json_loads(data):
            # on Python 3.5 json.loads only supports str not bytes
            return json.loads(data.decode())

def json(self):
        return ujson.loads(self.text)

    # Writes content into a file. This function will write while receiving, which avoids
    # having to load all content into memory

def deserialize(value):
        return loads(value)

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

def parse(self, method, data):
        try:
            json_data = json.loads(data)
            return json_data
        except Exception:
            raise AnApiOfIceAndFireError('Failed to parse JSON data')

def _parse(json_data):
        """Try and parse the JSON returned from Telegram.

        Returns:
            dict: A JSON parsed as Python dict with results - on error this dict will be empty.

        """
        decoded_s = json_data.decode('utf-8')
        try:
            data = json.loads(decoded_s)
        except ValueError:
            raise TelegramError('Invalid server response')

        if not data.get('ok'):
            description = data.get('description')
            parameters = data.get('parameters')
            if parameters:
                migrate_to_chat_id = parameters.get('migrate_to_chat_id')
                if migrate_to_chat_id:
                    raise ChatMigrated(migrate_to_chat_id)
                retry_after = parameters.get('retry_after')
                if retry_after:
                    raise RetryAfter(retry_after)
            if description:
                return description

        return data['result']

def _create_engine(self, config, section, all_pool_config):
        config_items = config[section]
        pool_policy = config[section]["pool_policy"]
        kws = {}
        if pool_policy is not None:
            pool_config = all_pool_config[pool_policy]
            if pool_config["poolclass"] != NullPool:
                kws = pool_config
        return create_engine(
            config_items["connect_url"],
            encoding=config_items["encoding"],
            connect_args=ujson.loads(config_items["connect_args"]),
            ** kws
        )