• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号

Python strings.utf82unicode函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中pyLibrary.strings.utf82unicode函数的典型用法代码示例。如果您正苦于以下问题:Python utf82unicode函数的具体用法?Python utf82unicode怎么用?Python utf82unicode使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



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

示例1: set_refresh_interval

    def set_refresh_interval(self, seconds):
        if seconds <= 0:
            interval = -1
        else:
            interval = unicode(seconds) + "s"

        if self.cluster.version.startswith("0.90."):
            response = self.cluster.put(
                "/" + self.settings.index + "/_settings",
                data='{"index":{"refresh_interval":' + convert.value2json(interval) + '}}'
            )

            result = convert.json2value(utf82unicode(response.all_content))
            if not result.ok:
                Log.error("Can not set refresh interval ({{error}})", {
                    "error": utf82unicode(response.all_content)
                })
        elif any(map(self.cluster.version.startswith, ["1.4.", "1.5.", "1.6.", "1.7."])):
            response = self.cluster.put(
                "/" + self.settings.index + "/_settings",
                data=convert.unicode2utf8('{"index":{"refresh_interval":' + convert.value2json(interval) + '}}')
            )

            result = convert.json2value(utf82unicode(response.all_content))
            if not result.acknowledged:
                Log.error("Can not set refresh interval ({{error}})", {
                    "error": utf82unicode(response.all_content)
                })
        else:
            Log.error("Do not know how to handle ES version {{version}}", version=self.cluster.version)
开发者ID:klahnakoski,项目名称:MoDataSubmission,代码行数:30,代码来源:elasticsearch.py


示例2: get

 def get(self, path, **kwargs):
     url = self.settings.host + ":" + unicode(self.settings.port) + path
     try:
         response = http.get(url, **kwargs)
         if response.status_code not in [200]:
             Log.error(response.reason+": "+response.all_content)
         if self.debug:
             Log.note("response: {{response}}", response=strings.limit(utf82unicode(response.all_content), 130))
         details = wrap(convert.json2value(utf82unicode(response.all_content)))
         if details.error:
             Log.error(details.error)
         return details
     except Exception, e:
         Log.error("Problem with call to {{url}}", url=url, cause=e)
开发者ID:klahnakoski,项目名称:MoDataSubmission,代码行数:14,代码来源:elasticsearch.py


示例3: post

    def post(self, path, **kwargs):
        url = self.settings.host + ":" + unicode(self.settings.port) + path

        try:
            wrap(kwargs).headers["Accept-Encoding"] = "gzip,deflate"

            data = kwargs.get(b'data')
            if data == None:
                pass
            elif isinstance(data, Mapping):
                kwargs[b'data'] = data =convert.unicode2utf8(convert.value2json(data))
            elif not isinstance(kwargs["data"], str):
                Log.error("data must be utf8 encoded string")

            if self.debug:
                sample = kwargs.get(b'data', "")[:300]
                Log.note("{{url}}:\n{{data|indent}}", url=url, data=sample)

            if self.debug:
                Log.note("POST {{url}}", url=url)
            response = http.post(url, **kwargs)
            if response.status_code not in [200, 201]:
                Log.error(response.reason.decode("latin1") + ": " + strings.limit(response.content.decode("latin1"), 100 if self.debug else 10000))
            if self.debug:
                Log.note("response: {{response}}", response=utf82unicode(response.content)[:130])
            details = convert.json2value(utf82unicode(response.content))
            if details.error:
                Log.error(convert.quote2string(details.error))
            if details._shards.failed > 0:
                Log.error("Shard failures {{failures|indent}}",
                    failures="---\n".join(r.replace(";", ";\n") for r in details._shards.failures.reason)
                )
            return details
        except Exception, e:
            if url[0:4] != "http":
                suggestion = " (did you forget \"http://\" prefix on the host name?)"
            else:
                suggestion = ""

            if kwargs.get("data"):
                Log.error(
                    "Problem with call to {{url}}" + suggestion + "\n{{body|left(10000)}}",
                    url=url,
                    body=strings.limit(kwargs["data"], 100 if self.debug else 10000),
                    cause=e
                )
            else:
                Log.error("Problem with call to {{url}}" + suggestion, url=url, cause=e)
开发者ID:klahnakoski,项目名称:esReplicate,代码行数:48,代码来源:elasticsearch.py


示例4: _typed_encode

def _typed_encode(value, _buffer):
    try:
        if value is None:
            append(_buffer, u'{"$value": null}')
            return
        elif value is True:
            append(_buffer, u'{"$value": true}')
            return
        elif value is False:
            append(_buffer, u'{"$value": false}')
            return

        _type = value.__class__
        if _type in (dict, Dict):
            if value:
                _dict2json(value, _buffer)
            else:
                append(_buffer, u'{"$object": "."}')
        elif _type is str:
            append(_buffer, u'{"$value": "')
            try:
                v = utf82unicode(value)
            except Exception, e:
                problem_serializing(value, e)

            for c in v:
                append(_buffer, ESCAPE_DCT.get(c, c))
            append(_buffer, u'"}')
        elif _type is unicode:
            append(_buffer, u'{"$value": "')
            for c in value:
                append(_buffer, ESCAPE_DCT.get(c, c))
            append(_buffer, u'"}')
开发者ID:klahnakoski,项目名称:TestFailures,代码行数:33,代码来源:typed_encoder.py


示例5: _value2json

def _value2json(value, _buffer):
    try:
        if value is None:
            append(_buffer, "null")
            return
        elif value is True:
            append(_buffer, "true")
            return
        elif value is False:
            append(_buffer, "false")
            return
        elif isinstance(value, Mapping):
            if value:
                _dict2json(value, _buffer)
            else:
                append(_buffer, "{}")
            return

        type = value.__class__
        if type is str:
            append(_buffer, '"')
            try:
                v = utf82unicode(value)
            except Exception, e:
                problem_serializing(value, e)

            for c in v:
                append(_buffer, ESCAPE_DCT.get(c, c))
            append(_buffer, '"')
        elif type is unicode:
            append(_buffer, '"')
            for c in value:
                append(_buffer, ESCAPE_DCT.get(c, c))
            append(_buffer, '"')
开发者ID:klahnakoski,项目名称:MoDataSubmission,代码行数:34,代码来源:encoder.py


示例6: _value2json

def _value2json(value, _buffer):
    try:
        _class = value.__class__
        if value is None:
            append(_buffer, u"null")
            return
        elif value is True:
            append(_buffer, u"true")
            return
        elif value is False:
            append(_buffer, u"false")
            return

        type = value.__class__
        if type is str:
            append(_buffer, u"\"")
            try:
                v = utf82unicode(value)
            except Exception, e:
                problem_serializing(value, e)

            for c in v:
                append(_buffer, ESCAPE_DCT.get(c, c))
            append(_buffer, u"\"")
        elif type is unicode:
            append(_buffer, u"\"")
            for c in value:
                append(_buffer, ESCAPE_DCT.get(c, c))
            append(_buffer, u"\"")
开发者ID:klahnakoski,项目名称:TestFailures,代码行数:29,代码来源:encoder.py


示例7: delete_index

    def delete_index(self, index_name):
        if not isinstance(index_name, unicode):
            Log.error("expecting an index name")

        if self.debug:
            Log.note("Deleting index {{index}}", index=index_name)

        # REMOVE ALL ALIASES TOO
        aliases = [a for a in self.get_aliases() if a.index == index_name and a.alias != None]
        if aliases:
            self.post(
                path="/_aliases",
                data={"actions": [{"remove": a} for a in aliases]}
            )

        url = self.settings.host + ":" + unicode(self.settings.port) + "/" + index_name
        try:
            response = http.delete(url)
            if response.status_code != 200:
                Log.error("Expecting a 200")
            details = convert.json2value(utf82unicode(response.content))
            if self.debug:
                Log.note("delete response {{response}}", response=details)
            return response
        except Exception, e:
            Log.error("Problem with call to {{url}}", url=url, cause=e)
开发者ID:klahnakoski,项目名称:esReplicate,代码行数:26,代码来源:elasticsearch.py


示例8: pretty_json

def pretty_json(value):
    try:
        if scrub(value) is None:
            return "null"
        elif isinstance(value, basestring):
            if isinstance(value, str):
                value = utf82unicode(value)
            try:
                return quote(value)
            except Exception, e:
                from pyLibrary.debugs.logs import Log

                try:
                    Log.note("try explicit convert of string with length {{length}}",  length= len(value))
                    acc = [u"\""]
                    for c in value:
                        try:
                            try:
                                c2 = ESCAPE_DCT[c]
                            except Exception, h:
                                c2 = c
                            c3 = unicode(c2)
                            acc.append(c3)
                        except BaseException, g:
                            pass
                            # Log.warning("odd character {{ord}} found in string.  Ignored.",  ord= ord(c)}, cause=g)
                    acc.append(u"\"")
                    output = u"".join(acc)
                    Log.note("return value of length {{length}}",  length= len(output))
                    return output
开发者ID:klahnakoski,项目名称:intermittents,代码行数:30,代码来源:encoder.py


示例9: head

 def head(self, path, **kwargs):
     url = self.settings.host + ":" + unicode(self.settings.port) + path
     try:
         response = http.head(url, **kwargs)
         if response.status_code not in [200]:
             Log.error(response.reason+": "+response.content)
         if self.debug:
             Log.note("response: {{response}}",  response= utf82unicode(response.content)[:130])
         if response.content:
             details = wrap(convert.json2value(utf82unicode(response.content)))
             if details.error:
                 Log.error(details.error)
             return details
         else:
             return None  # WE DO NOT EXPECT content WITH HEAD REQUEST
     except Exception, e:
         Log.error("Problem with call to {{url}}",  url= url, cause=e)
开发者ID:klahnakoski,项目名称:intermittents,代码行数:17,代码来源:elasticsearch.py


示例10: delete

 def delete(self, path, **kwargs):
     url = self.settings.host + ":" + unicode(self.settings.port) + path
     try:
         response = convert.json2value(utf82unicode(http.delete(url, **kwargs).content))
         if self.debug:
             Log.note("delete response {{response}}",  response= response)
         return response
     except Exception, e:
         Log.error("Problem with call to {{url}}",  url= url, cause=e)
开发者ID:klahnakoski,项目名称:intermittents,代码行数:9,代码来源:elasticsearch.py


示例11: _dict2json

def _dict2json(value, _buffer):
    prefix = u"{\""
    for k, v in value.iteritems():
        append(_buffer, prefix)
        prefix = u", \""
        if isinstance(k, str):
            k = utf82unicode(k)
        for c in k:
            append(_buffer, ESCAPE_DCT.get(c, c))
        append(_buffer, u"\": ")
        _value2json(v, _buffer)
    append(_buffer, u"}")
开发者ID:klahnakoski,项目名称:intermittents,代码行数:12,代码来源:encoder.py


示例12: _post

    def _post(self, path, **kwargs):
        url = self.settings.host + ":" + unicode(self.settings.port) + path

        try:
            wrap(kwargs).headers["Accept-Encoding"] = "gzip,deflate"

            if "data" in kwargs and not isinstance(kwargs["data"], str):
                Log.error("data must be utf8 encoded string")

            if self.debug:
                sample = kwargs.get("data", "")[:300]
                Log.note("{{url}}:\n{{data|indent}}",  url= url,  data= sample)

            response = http.post(url, **kwargs)
            if response.status_code not in [200, 201]:
                Log.error(response.reason+": "+response.content)
            if self.debug:
                Log.note("response: {{response}}",  response= utf82unicode(response.content)[:130])
            details = convert.json2value(utf82unicode(response.content))
            if details.error:
                Log.error(convert.quote2string(details.error))
            if details._shards.failed > 0:
                Log.error("Shard failures {{failures|indent}}",
                    failures= "---\n".join(r.replace(";", ";\n") for r in details._shards.failures.reason)
                )
            return details
        except Exception, e:
            if url[0:4] != "http":
                suggestion = " (did you forget \"http://\" prefix on the host name?)"
            else:
                suggestion = ""

            if kwargs.get("data"):
                Log.error("Problem with call to {{url}}" + suggestion + "\n{{body|left(10000)}}",
                    url= url,
                    body= kwargs["data"][0:10000] if self.debug else kwargs["data"][0:100], cause=e)
            else:
                Log.error("Problem with call to {{url}}" + suggestion, {"url": url}, e)
开发者ID:klahnakoski,项目名称:intermittents,代码行数:38,代码来源:elasticsearch.py


示例13: _dict2json

def _dict2json(value, _buffer):
    prefix = u'{"$object": ".", "'
    for k, v in value.iteritems():
        append(_buffer, prefix)
        prefix = u", \""
        if isinstance(k, str):
            k = utf82unicode(k)
        if not isinstance(k, unicode):
            Log.error("Expecting property name to be a string")
        for c in k:
            append(_buffer, ESCAPE_DCT.get(c, c))
        append(_buffer, u"\": ")
        _typed_encode(v, _buffer)
    append(_buffer, u"}")
开发者ID:klahnakoski,项目名称:TestFailures,代码行数:14,代码来源:typed_encoder.py


示例14: output

        def output():
            try:
                path = self._filename
                if path.startswith("~"):
                    home_path = os.path.expanduser("~")
                    path = home_path + path[1::]

                with io.open(path, "rb") as f:
                    for line in f:
                        yield utf82unicode(line)
            except Exception, e:
                from pyLibrary.debugs.logs import Log

                Log.error("Can not read line from {{filename}}",  filename= self._filename, cause=e)
开发者ID:klahnakoski,项目名称:MoTreeherder,代码行数:14,代码来源:files.py


示例15: put

    def put(self, path, **kwargs):
        url = self.settings.host + ":" + unicode(self.settings.port) + path

        if self.debug:
            sample = kwargs["data"][:300]
            Log.note("PUT {{url}}:\n{{data|indent}}",  url= url,  data= sample)
        try:
            response = http.put(url, **kwargs)
            if response.status_code not in [200]:
                Log.error(response.reason+": "+response.all_content)
            if self.debug:
                Log.note("response: {{response}}",  response= utf82unicode(response.all_content)[0:300:])
            return response
        except Exception, e:
            Log.error("Problem with call to {{url}}",  url= url, cause=e)
开发者ID:klahnakoski,项目名称:MoDataSubmission,代码行数:15,代码来源:elasticsearch.py


示例16: _dict2json

def _dict2json(value, _buffer):
    try:
        prefix = '{"'
        for k, v in value.iteritems():
            append(_buffer, prefix)
            prefix = ', "'
            if isinstance(k, str):
                k = utf82unicode(k)
            for c in k:
                append(_buffer, ESCAPE_DCT.get(c, c))
            append(_buffer, '": ')
            _value2json(v, _buffer)
        append(_buffer, "}")
    except Exception, e:
        from pyLibrary.debugs.logs import Log

        Log.error(_repr(value) + " is not JSON serializable", cause=e)
开发者ID:klahnakoski,项目名称:MoDataSubmission,代码行数:17,代码来源:encoder.py


示例17: sorted

                items = sorted(items, lambda a, b: value_compare(a[0], b[0]))
                values = [unicode_key(k) + ": " + indent(pretty_json(v)).strip() for k, v in items if v != None]
                return "{\n" + INDENT + (",\n" + INDENT).join(values) + "\n}"
            except Exception, e:
                from pyLibrary.debugs.logs import Log
                from pyLibrary.collections import OR

                if OR(not isinstance(k, basestring) for k in value.keys()):
                    Log.error("JSON must have string keys: {{keys}}:", keys=[k for k in value.keys()], cause=e)

                Log.error("problem making dict pretty: keys={{keys}}:", keys=[k for k in value.keys()], cause=e)
        elif value in (None, Null):
            return "null"
        elif isinstance(value, basestring):
            if isinstance(value, str):
                value = utf82unicode(value)
            try:
                return quote(value)
            except Exception, e:
                from pyLibrary.debugs.logs import Log

                try:
                    Log.note("try explicit convert of string with length {{length}}", length=len(value))
                    acc = ['"']
                    for c in value:
                        try:
                            try:
                                c2 = ESCAPE_DCT[c]
                            except Exception, h:
                                c2 = c
                            c3 = unicode(c2)
开发者ID:klahnakoski,项目名称:MoDataSubmission,代码行数:31,代码来源:encoder.py



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


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python threads.Queue类代码示例发布时间:2022-05-25
下一篇:
Python strings.expand_template函数代码示例发布时间:2022-05-25
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap