本文整理汇总了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;未经允许,请勿转载。 |
请发表评论