Read a column descriptor packet for each column in the result.
(self)
| 1446 | return tuple(row) |
| 1447 | |
| 1448 | def _get_descriptions(self): |
| 1449 | """Read a column descriptor packet for each column in the result.""" |
| 1450 | self.fields = [] |
| 1451 | self.converters = [] |
| 1452 | use_unicode = self.connection.use_unicode |
| 1453 | conn_encoding = self.connection.encoding |
| 1454 | description = [] |
| 1455 | |
| 1456 | for i in range(self.field_count): |
| 1457 | field = self.connection._read_packet(FieldDescriptorPacket) |
| 1458 | self.fields.append(field) |
| 1459 | description.append(field.description()) |
| 1460 | field_type = field.type_code |
| 1461 | if use_unicode: |
| 1462 | if field_type == FIELD_TYPE.JSON: |
| 1463 | # When SELECT from JSON column: charset = binary |
| 1464 | # When SELECT CAST(... AS JSON): charset = connection encoding |
| 1465 | # This behavior is different from TEXT / BLOB. |
| 1466 | # We should decode result by connection encoding regardless charsetnr. |
| 1467 | # See https://github.com/PyMySQL/PyMySQL/issues/488 |
| 1468 | encoding = conn_encoding # SELECT CAST(... AS JSON) |
| 1469 | elif field_type in TEXT_TYPES: |
| 1470 | if field.charsetnr == 63: # binary |
| 1471 | # TEXTs with charset=binary means BINARY types. |
| 1472 | encoding = None |
| 1473 | else: |
| 1474 | encoding = conn_encoding |
| 1475 | else: |
| 1476 | # Integers, Dates and Times, and other basic data is encoded in ascii |
| 1477 | encoding = "ascii" |
| 1478 | else: |
| 1479 | encoding = None |
| 1480 | converter = self.connection.decoders.get(field_type) |
| 1481 | if converter is converters.through: |
| 1482 | converter = None |
| 1483 | if DEBUG: |
| 1484 | print(f"DEBUG: field={field}, converter={converter}") |
| 1485 | self.converters.append((encoding, converter)) |
| 1486 | |
| 1487 | eof_packet = self.connection._read_packet() |
| 1488 | assert eof_packet.is_eof_packet(), "Protocol error, expecting EOF" |
| 1489 | self.description = tuple(description) |
| 1490 | |
| 1491 | |
| 1492 | def _send_local_file(filename: str, conn: Connection): |
no test coverage detected