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

Python utils.read_lc_string函数代码示例

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

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



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

示例1: parse_handshake_response

    def parse_handshake_response(packet):
        """Parse the handshake response packet and returns the information.

        :param packet: Packet received as received from socket without header.
        :return: Dictionary containing the handshake information; empty if
                 there was an error.
        :rtype: dict
        """
        try:
            (capabilities, max_allowed_packet, charset) = struct.unpack_from(
                "<IIB23x", buffer(packet[0:32]))
            buf = packet[32:]
            (buf, username) = utils.read_string(buf, end='\x00')
            (buf, auth_data) = utils.read_lc_string(buf)
            if auth_data and auth_data[-1] == '\x00':
                    auth_data = auth_data[:-1]
            if capabilities & ClientFlag.CONNECT_WITH_DB and buf:
                (buf, database) = utils.read_string(buf, end='\x00')
        except Exception as exc:
            _LOGGER.debug("Failed parsing handshake: %s", str(exc))
            return {}

        return {
            'capabilities': capabilities,
            'max_allowed_packet': max_allowed_packet,
            'charset': charset,
            'username': str(username),
            'auth_data': auth_data,
        }
开发者ID:dzolnierz,项目名称:mysql-utilities,代码行数:29,代码来源:mysqlrpc.py


示例2: _parse_binary_values

    def _parse_binary_values(self, fields, packet):
        """Parse values from a binary result packet"""
        null_bitmap_length = (len(fields) + 7 + 2) // 8
        null_bitmap = [int(i) for i in packet[0:null_bitmap_length]]
        packet = packet[null_bitmap_length:]

        values = []
        for pos, field in enumerate(fields):
            if null_bitmap[int((pos+2)/8)] & (1 << (pos + 2) % 8):
                values.append(None)
                continue
            elif field[1] in (FieldType.TINY, FieldType.SHORT,
                              FieldType.INT24,
                              FieldType.LONG, FieldType.LONGLONG):
                (packet, value) = self._parse_binary_integer(packet, field)
                values.append(value)
            elif field[1] in (FieldType.DOUBLE, FieldType.FLOAT):
                (packet, value) = self._parse_binary_float(packet, field)
                values.append(value)
            elif field[1] in (FieldType.DATETIME, FieldType.DATE,
                              FieldType.TIMESTAMP):
                (packet, value) = self._parse_binary_timestamp(packet, field)
                values.append(value)
            elif field[1] == FieldType.TIME:
                (packet, value) = self._parse_binary_time(packet, field)
                values.append(value)
            else:
                (packet, value) = utils.read_lc_string(packet)
                values.append(value)

        return tuple(values)
开发者ID:richfab,项目名称:LOG210-server,代码行数:31,代码来源:protocol.py


示例3: test_read_lc_string_5

 def test_read_lc_string_5(self):
     """Read a length code string from a buffer which is 'NULL'"""
     exp = 'abc'
     lcs = '\xfb' + 'abc'
     
     (rest, result) = utils.read_lc_string(lcs)
     if result != None or rest != 'abc':
         self.fail("Wrong result. Expected None.")
开发者ID:BugZuo,项目名称:awesome-python-webapp,代码行数:8,代码来源:test_utils.py


示例4: test_read_lc_string_5

    def test_read_lc_string_5(self):
        """Read a length code string from a buffer which is 'NULL'"""
        exp = bytearray(b'abc')
        lcs = bytearray(b'\xfb') + exp

        (rest, result) = utils.read_lc_string(lcs)
        if result != None or rest != exp:
            self.fail("Wrong result. Expected None.")
开发者ID:buildfax,项目名称:mysql-connector-python,代码行数:8,代码来源:test_utils.py


示例5: test_read_lc_string_8

    def test_read_lc_string_8(self):
        """Read a length code string from a buffer ( <= 2^32 bytes)"""
        exp = "a" * (2**24)
        expsize = len(exp)
        lcs = '\xfe' + utils.int8store(expsize) + exp

        (rest, result) = utils.read_lc_string(lcs)
        if result != exp or len(result) != expsize:
            self.fail("Wrong result. Expected size '%d', got '%d'" %\
                expsize, len(result))
开发者ID:BugZuo,项目名称:awesome-python-webapp,代码行数:10,代码来源:test_utils.py


示例6: test_read_lc_string_2

    def test_read_lc_string_2(self):
        """Read a length code string from a buffer ( <= 2^16 bytes)"""
        exp = "a" * 2**(16-1)
        expsize = len(exp)
        lcs = '\xfc' + utils.int2store(expsize) + exp

        (rest, result) = utils.read_lc_string(lcs)
        if result != exp or len(result) != expsize:
            self.fail("Wrong result. Expected '%d', got '%d'" %\
                expsize, len(result))
开发者ID:BugZuo,项目名称:awesome-python-webapp,代码行数:10,代码来源:test_utils.py


示例7: test_read_lc_string_8

    def test_read_lc_string_8(self):
        """Read a length code string from a buffer ( <= 2^64 bytes)"""
        exp = bytearray(b"a" * 2 ** 24)
        expsize = len(exp)
        lcs = bytearray(b'\xfe') + utils.int8store(expsize) + exp

        (_, result) = utils.read_lc_string(lcs)
        if result != exp or len(result) != expsize:
            self.fail("Wrong result. Expected '{0}', got '{1}'".format(
                expsize, len(result)))
开发者ID:buildfax,项目名称:mysql-connector-python,代码行数:10,代码来源:test_utils.py


示例8: test_read_lc_string_3

    def test_read_lc_string_3(self):
        """Read a length code string from a buffer ( <= 2^24 bytes)"""
        exp = "a" * 2 ** (24 - 1)
        expsize = len(exp)
        lcs = '\xfd' + utils.int3store(expsize) + exp

        (_, result) = utils.read_lc_string(lcs)
        if result != exp or len(result) != expsize:
            self.fail("Wrong result. Expected '{0}', got '{1}'".format(
                expsize, len(result)))
开发者ID:allienson,项目名称:LP_trab_python,代码行数:10,代码来源:test_utils.py


示例9: test_read_lc_string_3

    def test_read_lc_string_3(self):
        """Read a length code string from a buffer ( <= 2^24 bytes)"""
        exp = b"a" * 2**(24-1)
        expsize = len(exp)
        lcs = b'\xfd' + utils.int3store(expsize) + exp

        (rest, result) = utils.read_lc_string(lcs)
        if result != exp or len(result) != expsize:
            self.fail("Wrong result. Expected size'%d', got '%d'" %\
                expsize, len(result))
开发者ID:BugZuo,项目名称:awesome-python-webapp,代码行数:10,代码来源:test_utils.py


示例10: parse_column

    def parse_column(self, packet):
        """Parse a MySQL column-packet"""
        (packet, _) = utils.read_lc_string(packet[4:]) # catalog
        (packet, _) = utils.read_lc_string(packet)  # db
        (packet, _) = utils.read_lc_string(packet)  # table
        (packet, _) = utils.read_lc_string(packet)  # org_table
        (packet, name) = utils.read_lc_string(packet)  # name
        (packet, _) = utils.read_lc_string(packet)  # org_name

        try:
            (_, _, field_type,
             flags, _) = struct.unpack('<xHIBHBxx', packet)
        except struct.error:
            raise errors.InterfaceError("Failed parsing column information")

        return (
            name.decode('utf-8'),
            field_type,
            None,  # display_size
            None,  # internal_size
            None,  # precision
            None,  # scale
            ~flags & FieldFlag.NOT_NULL,  # null_ok
            flags,  # MySQL specific
            )
开发者ID:richfab,项目名称:LOG210-server,代码行数:25,代码来源:protocol.py


示例11: parse_column

 def parse_column(self, packet):
     """Parse a MySQL column-packet"""
     column = {}
     (packet, column['catalog']) = utils.read_lc_string(packet[4:])
     (packet, column['db']) = utils.read_lc_string(packet)
     (packet, column['table']) = utils.read_lc_string(packet)
     (packet, column['org_table']) = utils.read_lc_string(packet)
     (packet, column['name']) = utils.read_lc_string(packet)
     (packet, column['org_name']) = utils.read_lc_string(packet)
     packet = packet[1:] # filler 1 * \x00
     (packet, column['charset']) = utils.read_int(packet, 2)
     (packet, column['length']) = utils.read_int(packet,4)
     (packet, column['type']) = utils.read_int(packet, 1)
     (packet, column['flags']) = utils.read_int(packet, 2)
     (packet, column['decimal']) = utils.read_int(packet, 1)
     packet = packet[2:] # filler 2 * \x00
     
     return (
         column['name'].decode('utf-8'),
         column['type'],
         None, # display_size
         None, # internal_size
         None, # precision
         None, # scale
         ~column['flags'] & FieldFlag.NOT_NULL, # null_ok
         column['flags'], # MySQL specific
         )
开发者ID:hwangsyin,项目名称:cbrc-devteam-blog,代码行数:27,代码来源:protocol.py


示例12: parse_ok

    def parse_ok(self, packet):
        """Parse a MySQL OK-packet"""
        if not packet[4] == '\x00':
            raise errors.InterfaceError("Failed parsing OK packet.")

        ok = {}
        try:
            (packet, ok['field_count']) = utils.read_int(packet[4:], 1)
            (packet, ok['affected_rows']) = utils.read_lc_int(packet)
            (packet, ok['insert_id']) = utils.read_lc_int(packet)
            (packet, ok['server_status']) = utils.read_int(packet, 2)
            (packet, ok['warning_count']) = utils.read_int(packet, 2)
            if packet:
                (packet, ok['info_msg']) = utils.read_lc_string(packet)
        except ValueError:
            raise errors.InterfaceError("Failed parsing OK packet.")
        return ok
开发者ID:khankuan,项目名称:IvleHangout,代码行数:17,代码来源:protocol.py


示例13: parse_ok

    def parse_ok(self, packet):
        """Parse a MySQL OK-packet"""
        if not packet[4] == '\x00':
            raise errors.InterfaceError("Failed parsing OK packet.")

        ok_packet = {}
        try:
            ok_packet['field_count'] = struct.unpack('<xxxxB', packet[0:5])[0]
            (packet, ok_packet['affected_rows']) = utils.read_lc_int(packet[5:])
            (packet, ok_packet['insert_id']) = utils.read_lc_int(packet)
            (ok_packet['server_status'],
                ok_packet['warning_count']) = struct.unpack('<HH', packet[0:4])
            packet = packet[4:]
            if packet:
                (packet, ok_packet['info_msg']) = utils.read_lc_string(packet)
        except ValueError:
            raise errors.InterfaceError("Failed parsing OK packet.")
        return ok_packet
开发者ID:christopherfan,项目名称:urlShorty,代码行数:18,代码来源:protocol.py


示例14: parse_ok

    def parse_ok(self, packet):
        """Parse a MySQL OK-packet"""
        if not packet[4] == 0:
            raise errors.InterfaceError("Failed parsing OK packet.")

        ok_packet = {}
        try:
            ok_packet["field_count"] = struct.unpack("<xxxxB", packet[0:5])[0]
            (packet, ok_packet["affected_rows"]) = utils.read_lc_int(packet[5:])
            (packet, ok_packet["insert_id"]) = utils.read_lc_int(packet)
            (ok_packet["server_status"], ok_packet["warning_count"]) = struct.unpack("<HH", packet[0:4])
            packet = packet[4:]
            if packet:
                (packet, ok_packet["info_msg"]) = utils.read_lc_string(packet)
                ok_packet["info_msg"] = ok_packet["info_msg"].decode("utf-8")
        except ValueError:
            raise errors.InterfaceError("Failed parsing OK packet.")
        return ok_packet
开发者ID:rcsousa,项目名称:heroku-buildpack-python,代码行数:18,代码来源:protocol.py



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


鲜花

握手

雷人

路过

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

请发表评论

全部评论

专题导读
上一篇:
Python events.trigger函数代码示例发布时间:2022-05-27
下一篇:
Python utils.int4store函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

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

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

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