# Copyright (c) 2016, 2022, Oracle and/or its affiliates.
|
#
|
# This program is free software; you can redistribute it and/or modify
|
# it under the terms of the GNU General Public License, version 2.0, as
|
# published by the Free Software Foundation.
|
#
|
# This program is also distributed with certain software (including
|
# but not limited to OpenSSL) that is licensed under separate terms,
|
# as designated in a particular file or component or in included license
|
# documentation. The authors of MySQL hereby grant you an
|
# additional permission to link the program and your derivative works
|
# with the separately licensed software that they have included with
|
# MySQL.
|
#
|
# Without limiting anything contained in the foregoing, this file,
|
# which is part of MySQL Connector/Python, is also subject to the
|
# Universal FOSS Exception, version 1.0, a copy of which can be found at
|
# http://oss.oracle.com/licenses/universal-foss-exception.
|
#
|
# This program is distributed in the hope that it will be useful, but
|
# WITHOUT ANY WARRANTY; without even the implied warranty of
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
|
# See the GNU General Public License, version 2.0, for more details.
|
#
|
# You should have received a copy of the GNU General Public License
|
# along with this program; if not, write to the Free Software Foundation, Inc.,
|
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
|
|
"""Implementation of the Result classes."""
|
|
from __future__ import annotations
|
|
import decimal
|
import struct
|
import sys
|
|
from datetime import datetime, timedelta
|
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
|
|
from .charsets import MYSQL_CHARACTER_SETS
|
from .dbdoc import DbDoc
|
from .helpers import decode_from_bytes, deprecated
|
from .types import ConnectionType, FieldTypes
|
|
|
# pylint: disable=missing-class-docstring,missing-function-docstring
|
def from_protobuf(column: Column, payload: bytes) -> Any:
|
if len(payload) == 0:
|
return None
|
|
if column.get_type() == ColumnType.STRING:
|
return decode_from_bytes(payload[:-1]) # Strip trailing char
|
|
try:
|
return ColumnProtoType.converter_map[column.get_proto_type()](payload)
|
except KeyError as err:
|
sys.stderr.write(f"{err}")
|
sys.stderr.write(f"{payload.encode('hex')}") # type: ignore[attr-defined]
|
return None
|
|
|
def bytes_from_protobuf(payload: bytes) -> bytes:
|
# Strip trailing char
|
return payload[:-1]
|
|
|
def float_from_protobuf(payload: bytes) -> float:
|
assert len(payload) == 4
|
return struct.unpack("<f", payload)[0]
|
|
|
def double_from_protobuf(payload: bytes) -> float:
|
assert len(payload) == 8
|
return struct.unpack("<d", payload)[0]
|
|
|
def varint_from_protobuf_stream(payload: bytes) -> Tuple[int, bytes]:
|
if len(payload) == 0:
|
raise ValueError("Payload is empty")
|
|
cur = 0
|
i = 0
|
shift = 0
|
|
for item in payload:
|
char = item if isinstance(item, int) else ord(item) # type: ignore[arg-type]
|
eos = (char & 0x80) == 0
|
cur_bits = char & 0x7F
|
cur_bits <<= shift
|
i |= cur_bits
|
if eos:
|
return i, payload[cur + 1 :]
|
cur += 1
|
shift += 7
|
|
raise EOFError("Payload too short")
|
|
|
def varint_from_protobuf(payload: bytes) -> int:
|
i, payload = varint_from_protobuf_stream(payload)
|
if len(payload) != 0:
|
raise ValueError("Payload too long")
|
|
return i
|
|
|
def varsint_from_protobuf(payload: bytes) -> int:
|
i, payload = varint_from_protobuf_stream(payload)
|
if len(payload) != 0:
|
raise ValueError("Payload too long")
|
|
# Zigzag encoded, revert it
|
if i & 0x1:
|
i = ~i
|
i = i >> 1
|
i |= 1 << 63
|
else:
|
i = i >> 1
|
|
return i
|
|
|
def set_from_protobuf(payload: bytes) -> List[bytes]:
|
set_pb: List = []
|
while True:
|
try:
|
field_len, payload = varint_from_protobuf_stream(payload)
|
if len(payload) < field_len:
|
if len(payload) == 0 and field_len == 1 and len(set_pb) == 0:
|
# Special case for empty set
|
return []
|
raise ValueError("Invalid Set encoding")
|
|
set_pb.append(payload[:field_len])
|
payload = payload[field_len:]
|
if len(payload) == 0:
|
# Done
|
break
|
except ValueError:
|
break
|
return set_pb
|
|
|
def decimal_from_protobuf(payload: bytes) -> decimal.Decimal:
|
digits = []
|
sign = None
|
scale = payload[0] if isinstance(payload[0], int) else ord(payload[0]) # type: ignore[arg-type]
|
payload = payload[1:]
|
|
for item in payload:
|
char = item if isinstance(item, int) else ord(item) # type: ignore[arg-type]
|
high_bcd = (char & 0xF0) >> 4
|
low_bcd = char & 0x0F
|
if high_bcd < 0x0A:
|
digits.append(high_bcd)
|
if low_bcd < 0x0A:
|
digits.append(low_bcd)
|
elif low_bcd == 0x0C:
|
sign = 0
|
break
|
elif low_bcd == 0x0D:
|
sign = 1
|
break
|
else:
|
raise ValueError("Invalid BCD")
|
elif high_bcd == 0x0C:
|
sign = 0
|
assert low_bcd == 0x00
|
break
|
elif high_bcd == 0x0D:
|
sign = 1
|
assert low_bcd == 0x00
|
break
|
else:
|
raise ValueError(f"Invalid BCD: {high_bcd}")
|
|
return decimal.Decimal((sign, digits, -scale))
|
|
|
def datetime_from_protobuf(payload: bytes) -> datetime:
|
# A sequence of varints
|
hour = 0
|
minutes = 0
|
seconds = 0
|
useconds = 0
|
year, payload = varint_from_protobuf_stream(payload)
|
month, payload = varint_from_protobuf_stream(payload)
|
day, payload = varint_from_protobuf_stream(payload)
|
|
try:
|
hour, payload = varint_from_protobuf_stream(payload)
|
minutes, payload = varint_from_protobuf_stream(payload)
|
seconds, payload = varint_from_protobuf_stream(payload)
|
useconds, payload = varint_from_protobuf_stream(payload)
|
except ValueError:
|
pass
|
|
return datetime(year, month, day, hour, minutes, seconds, useconds)
|
|
|
def time_from_protobuf(payload: bytes) -> timedelta:
|
# A sequence of varints
|
hour = 0
|
minutes = 0
|
seconds = 0
|
useconds = 0
|
negate = payload[0] == 1
|
payload = payload[1:]
|
|
try:
|
hour, payload = varint_from_protobuf_stream(payload)
|
minutes, payload = varint_from_protobuf_stream(payload)
|
seconds, payload = varint_from_protobuf_stream(payload)
|
useconds, payload = varint_from_protobuf_stream(payload)
|
except ValueError:
|
pass
|
|
if negate:
|
# Negate the first non-zero value
|
if hour:
|
hour *= -1
|
elif minutes:
|
minutes *= -1
|
elif seconds:
|
seconds *= -1
|
elif useconds:
|
useconds *= -1
|
|
return timedelta(
|
hours=hour, minutes=minutes, seconds=seconds, microseconds=useconds
|
)
|
|
|
class Collations:
|
UTF8_GENERAL_CI = 33
|
|
|
class ColumnType:
|
BIT = 1
|
TINYINT = 2
|
SMALLINT = 3
|
MEDIUMINT = 4
|
INT = 5
|
BIGINT = 6
|
REAL = 7
|
FLOAT = 8
|
DECIMAL = 9
|
NUMERIC = 10
|
DOUBLE = 11
|
JSON = 12
|
STRING = 13
|
BYTES = 14
|
TIME = 15
|
DATE = 16
|
DATETIME = 17
|
TIMESTAMP = 18
|
SET = 19
|
ENUM = 20
|
GEOMETRY = 21
|
XML = 22
|
YEAR = 23
|
CHAR = 24
|
VARCHAR = 25
|
BINARY = 26
|
VARBINARY = 27
|
TINYBLOB = 28
|
BLOB = 29
|
MEDIUMBLOB = 30
|
LONGBLOB = 31
|
TINYTEXT = 32
|
TEXT = 33
|
MEDIUMTEXT = 34
|
LONGTEXT = 35
|
|
@classmethod
|
def to_string(cls, needle: Any) -> Optional[str]:
|
for key, value in vars(cls).items():
|
if value == needle:
|
return key
|
return None
|
|
@classmethod
|
def from_string(cls, key: str) -> Any:
|
return getattr(cls, key.upper(), None)
|
|
@classmethod
|
def is_char(cls, col_type: int) -> bool:
|
return col_type in (
|
cls.CHAR,
|
cls.VARCHAR,
|
)
|
|
@classmethod
|
def is_binary(cls, col_type: int) -> bool:
|
return col_type in (
|
cls.BINARY,
|
cls.VARBINARY,
|
)
|
|
@classmethod
|
def is_text(cls, col_type: int) -> bool:
|
return col_type in (
|
cls.TEXT,
|
cls.TINYTEXT,
|
cls.MEDIUMTEXT,
|
cls.LONGTEXT,
|
)
|
|
@classmethod
|
def is_decimals(cls, col_type: int) -> bool:
|
return col_type in (
|
cls.REAL,
|
cls.DOUBLE,
|
cls.FLOAT,
|
cls.DECIMAL,
|
cls.NUMERIC,
|
)
|
|
@classmethod
|
def is_numeric(cls, col_type: int) -> bool:
|
return col_type in (
|
cls.BIT,
|
cls.TINYINT,
|
cls.SMALLINT,
|
cls.MEDIUMINT,
|
cls.INT,
|
cls.BIGINT,
|
)
|
|
@classmethod
|
def is_finite_set(cls, col_type: int) -> bool:
|
return col_type in (
|
cls.SET,
|
cls.ENUM,
|
)
|
|
|
class ColumnProtoType:
|
SINT = 1
|
UINT = 2
|
DOUBLE = 5
|
FLOAT = 6
|
BYTES = 7
|
TIME = 10
|
DATETIME = 12
|
SET = 15
|
ENUM = 16
|
BIT = 17
|
DECIMAL = 18
|
|
converter_map: Dict[int, Callable[[bytes], Any]] = {
|
SINT: varsint_from_protobuf,
|
UINT: varint_from_protobuf,
|
BYTES: bytes_from_protobuf,
|
DATETIME: datetime_from_protobuf,
|
TIME: time_from_protobuf,
|
FLOAT: float_from_protobuf,
|
DOUBLE: double_from_protobuf,
|
BIT: varint_from_protobuf,
|
SET: set_from_protobuf,
|
ENUM: bytes_from_protobuf,
|
DECIMAL: decimal_from_protobuf,
|
}
|
|
|
class Flags:
|
def __init__(self, value: int) -> None:
|
self._allowed_flags: Dict[str, int] = {}
|
self._flag_names: Dict[int, str] = {}
|
for key, val in self.__class__.__dict__.items():
|
if key.startswith("__"):
|
continue
|
if isinstance(val, int):
|
self._allowed_flags[key] = val
|
self._flag_names[val] = key
|
self._value: int = value
|
|
def __str__(self) -> str:
|
mask = 1
|
flag_names = []
|
value = self._value
|
|
for _ in range(0, 63):
|
mask <<= 1
|
flag = value & mask
|
if flag:
|
# We matched something, find the name for it
|
try:
|
flag_names.append(self._flag_names[flag])
|
except KeyError:
|
sys.stderr.write(f"{self._flag_names}")
|
sys.stderr.write(f"{self.__class__.__dict__}")
|
|
return ",".join(flag_names)
|
|
@property
|
def value(self) -> int:
|
return self._value
|
|
@value.setter
|
def value(self, val: int) -> None:
|
self._value = val
|
|
|
class ColumnFlags(Flags):
|
NOT_NULL = 0x0010
|
PRIMARY_KEY = 0x0020
|
UNIQUE_KEY = 0x0040
|
MULTIPLE_KEY = 0x0080
|
AUTO_INCREMENT = 0x0100
|
|
|
class DatetimeColumnFlags(ColumnFlags):
|
TIMESTAMP = 0x0001
|
|
|
class UIntColumnFlags(ColumnFlags):
|
ZEROFILL = 0x0001
|
|
|
class DoubleColumnFlags(ColumnFlags):
|
UNSIGNED = 0x0001
|
|
|
class FloatColumnFlags(ColumnFlags):
|
UNSIGNED = 0x0001
|
|
|
class BytesColumnFlags(ColumnFlags):
|
RIGHT_PAD = 0x0001
|
|
|
class BytesContentType(ColumnFlags):
|
GEOMETRY = 0x0001
|
JSON = 0x0002
|
XML = 0x0003
|
|
|
# pylint: enable=missing-class-docstring,missing-function-docstring
|
|
|
class Column:
|
"""Represents meta data for a table column.
|
|
Args:
|
col_type (int): The column type.
|
catalog (str): The catalog.
|
schema (str): The schema name.
|
table (str): The table name.
|
original_table (str): The original table name.
|
name (str): The column name.
|
original_name (str): The original table name.
|
length (int): The column length,
|
collation (str): The collation name.
|
fractional_digits (int): The fractional digits.
|
flags (int): The flags.
|
content_type (int): The content type.
|
|
.. versionchanged:: 8.0.12
|
"""
|
|
def __init__(
|
self,
|
col_type: int,
|
catalog: Optional[str] = None,
|
schema: Optional[str] = None,
|
table: Optional[str] = None,
|
original_table: Optional[str] = None,
|
name: Optional[str] = None,
|
original_name: Optional[str] = None,
|
length: Optional[int] = None,
|
collation: Optional[int] = None,
|
fractional_digits: Optional[int] = None,
|
flags: Optional[int] = None,
|
content_type: Optional[int] = None,
|
) -> None:
|
self._schema: str = decode_from_bytes(schema)
|
self._name: str = decode_from_bytes(name)
|
self._original_name: str = decode_from_bytes(original_name)
|
self._table: str = decode_from_bytes(table)
|
self._original_table: str = decode_from_bytes(original_table)
|
self._proto_type: int = col_type
|
self._col_type: Optional[int] = None
|
self._catalog: Optional[str] = catalog
|
self._length: Optional[int] = length
|
self._collation: Optional[int] = collation
|
self._fractional_digits: Optional[int] = fractional_digits
|
self._flags: Optional[int] = flags
|
self._content_type: Optional[int] = content_type
|
self._number_signed: bool = False
|
self._is_padded: Union[bool, int] = False
|
self._is_binary: bool = False
|
self._is_bytes: bool = False
|
self._collation_name: Optional[str] = None
|
self._character_set_name: Optional[str] = None
|
self._zero_fill: Optional[int] = None
|
|
if self._collation > 0:
|
if self._collation >= len(MYSQL_CHARACTER_SETS):
|
raise ValueError(f"No mapping found for collation {self._collation}")
|
info = MYSQL_CHARACTER_SETS[self._collation]
|
self._character_set_name = info[0]
|
self._collation_name = info[1]
|
self._is_binary = (
|
"binary" in self._collation_name or "_bin" in self._collation_name
|
)
|
self._map_type()
|
self._is_bytes = self._col_type in (
|
ColumnType.GEOMETRY,
|
ColumnType.JSON,
|
ColumnType.XML,
|
ColumnType.BYTES,
|
ColumnType.STRING,
|
)
|
|
def __str__(self) -> str:
|
return str(
|
{
|
"col_type": self._col_type,
|
"schema": self._schema,
|
"table": self._table,
|
"flags": str(self._flags),
|
}
|
)
|
|
def _map_bytes(self) -> None:
|
"""Map bytes."""
|
if self._content_type == BytesContentType.GEOMETRY:
|
self._col_type = ColumnType.GEOMETRY
|
elif self._content_type == BytesContentType.JSON:
|
self._col_type = ColumnType.JSON
|
elif self._content_type == BytesContentType.XML:
|
self._col_type = ColumnType.XML
|
elif self._is_binary:
|
self._col_type = ColumnType.BYTES
|
else:
|
self._col_type = ColumnType.STRING
|
self._is_padded = self._flags & 1
|
|
def _map_datetime(self) -> None:
|
"""Map datetime."""
|
if self._length == 10:
|
self._col_type = ColumnType.DATE
|
elif self._flags & DatetimeColumnFlags.TIMESTAMP > 0:
|
self._col_type = ColumnType.TIMESTAMP
|
elif self._length >= 19:
|
self._col_type = ColumnType.DATETIME
|
else:
|
raise ValueError("Datetime mapping scenario unhandled")
|
|
def _map_int_type(self) -> None:
|
"""Map int type."""
|
if self._length <= 4:
|
self._col_type = ColumnType.TINYINT
|
elif self._length <= 6:
|
self._col_type = ColumnType.SMALLINT
|
elif self._length <= 9:
|
self._col_type = ColumnType.MEDIUMINT
|
elif self._length <= 11:
|
self._col_type = ColumnType.INT
|
else:
|
self._col_type = ColumnType.BIGINT
|
self._number_signed = True
|
|
def _map_uint_type(self) -> None:
|
"""Map uint type."""
|
if self._length <= 3:
|
self._col_type = ColumnType.TINYINT
|
elif self._length <= 5:
|
self._col_type = ColumnType.SMALLINT
|
elif self._length <= 8:
|
self._col_type = ColumnType.MEDIUMINT
|
elif self._length <= 10:
|
self._col_type = ColumnType.INT
|
else:
|
self._col_type = ColumnType.BIGINT
|
self._zero_fill = self._flags & 1
|
|
def _map_type(self) -> None:
|
"""Map type."""
|
if self._proto_type == ColumnProtoType.SINT:
|
self._map_int_type()
|
elif self._proto_type == ColumnProtoType.UINT:
|
self._map_uint_type()
|
elif self._proto_type == ColumnProtoType.FLOAT:
|
self._col_type = ColumnType.FLOAT
|
self._is_number_signed = (self._flags & FloatColumnFlags.UNSIGNED) == 0
|
elif self._proto_type == ColumnProtoType.DECIMAL:
|
self._col_type = ColumnType.DECIMAL
|
self._is_number_signed = (self._flags & FloatColumnFlags.UNSIGNED) == 0
|
elif self._proto_type == ColumnProtoType.DOUBLE:
|
self._col_type = ColumnType.DOUBLE
|
self._is_number_signed = (self._flags & FloatColumnFlags.UNSIGNED) == 0
|
elif self._proto_type == ColumnProtoType.BYTES:
|
self._map_bytes()
|
elif self._proto_type == ColumnProtoType.TIME:
|
self._col_type = ColumnType.TIME
|
elif self._proto_type == ColumnProtoType.DATETIME:
|
self._map_datetime()
|
elif self._proto_type == ColumnProtoType.SET:
|
self._col_type = ColumnType.SET
|
elif self._proto_type == ColumnProtoType.ENUM:
|
self._col_type = ColumnType.ENUM
|
elif self._proto_type == ColumnProtoType.BIT:
|
self._col_type = ColumnType.BIT
|
else:
|
raise ValueError(f"Unknown column type {self._proto_type}")
|
|
@property
|
def schema_name(self) -> str:
|
"""str: The schema name.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._schema
|
|
@property
|
def table_name(self) -> str:
|
"""str: The table name.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._original_table or self._table
|
|
@property
|
def table_label(self) -> str:
|
"""str: The table label.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._table or self._original_table
|
|
@property
|
def column_name(self) -> str:
|
"""str: The column name.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._original_name or self._name
|
|
@property
|
def column_label(self) -> str:
|
"""str: The column label.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._name or self._original_name
|
|
@property
|
def type(self) -> int:
|
"""int: The column type.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._col_type
|
|
@property
|
def length(self) -> int:
|
"""int. The column length.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._length
|
|
@property
|
def fractional_digits(self) -> int:
|
"""int: The column fractional digits.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._fractional_digits
|
|
@property
|
def collation_name(self) -> str:
|
"""str: The collation name.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._collation_name
|
|
@property
|
def character_set_name(self) -> str:
|
"""str: The character set name.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._character_set_name
|
|
def get_schema_name(self) -> str:
|
"""Returns the schema name.
|
|
Returns:
|
str: The schema name.
|
"""
|
return self._schema
|
|
def get_table_name(self) -> str:
|
"""Returns the table name.
|
|
Returns:
|
str: The table name.
|
"""
|
return self._original_table or self._table
|
|
def get_table_label(self) -> str:
|
"""Returns the table label.
|
|
Returns:
|
str: The table label.
|
"""
|
return self._table or self._original_table
|
|
def get_column_name(self) -> str:
|
"""Returns the column name.
|
|
Returns:
|
str: The column name.
|
"""
|
return self._original_name or self._name
|
|
def get_column_label(self) -> str:
|
"""Returns the column label.
|
|
Returns:
|
str: The column label.
|
"""
|
return self._name or self._original_name
|
|
def get_proto_type(self) -> int:
|
"""Returns the column proto type.
|
|
Returns:
|
int: The column proto type.
|
"""
|
return self._proto_type
|
|
def get_type(self) -> int:
|
"""Returns the column type.
|
|
Returns:
|
int: The column type.
|
"""
|
return self._col_type
|
|
def get_length(self) -> int:
|
"""Returns the column length.
|
|
Returns:
|
int: The column length.
|
"""
|
return self._length
|
|
def get_fractional_digits(self) -> int:
|
"""Returns the column fractional digits.
|
|
Returns:
|
int: The column fractional digits.
|
"""
|
return self._fractional_digits
|
|
def get_collation_name(self) -> str:
|
"""Returns the collation name.
|
|
Returns:
|
str: The collation name.
|
"""
|
return self._collation_name
|
|
def get_character_set_name(self) -> str:
|
"""Returns the character set name.
|
|
Returns:
|
str: The character set name.
|
"""
|
return self._character_set_name
|
|
def is_number_signed(self) -> bool:
|
"""Returns `True` if is a number signed.
|
|
Returns:
|
bool: Returns `True` if is a number signed.
|
"""
|
return self._number_signed
|
|
def is_padded(self) -> Union[bool, int]:
|
"""Returns `True` if is padded.
|
|
Returns:
|
bool: Returns `True` if is padded.
|
"""
|
return self._is_padded
|
|
def is_bytes(self) -> bool:
|
"""Returns `True` if is bytes.
|
|
Returns:
|
bool: Returns `True` if is bytes.
|
"""
|
return self._is_bytes
|
|
|
class Row:
|
"""Represents a row element returned from a SELECT query.
|
|
Args:
|
resultset (mysqlx.SqlResult or mysqlx.RowResult): The result set.
|
fields (`list`): The list of fields.
|
"""
|
|
def __init__(
|
self, resultset: Union[BufferingResult, RowResult], fields: Sequence[FieldTypes]
|
) -> None:
|
self._fields: Sequence[FieldTypes] = fields
|
self._resultset: Union[BufferingResult, RowResult] = resultset
|
|
def __repr__(self) -> str:
|
return repr(self._fields)
|
|
def __getitem__(self, index: Union[int, str]) -> Any:
|
"""Returns the value of a column by name or index.
|
|
.. versionchanged:: 8.0.12
|
"""
|
int_index = self._resultset.index_of(index) if isinstance(index, str) else index
|
if int_index == -1 and isinstance(index, str):
|
raise ValueError(f"Column name '{index}' not found")
|
if int_index >= len(self._fields) or int_index < 0:
|
raise IndexError("Index out of range")
|
return self._fields[int_index]
|
|
@deprecated("8.0.12")
|
def get_string(self, str_index: str) -> str:
|
"""Returns the value using the column name.
|
|
Args:
|
str_index (str): The column name.
|
|
.. deprecated:: 8.0.12
|
"""
|
int_index = self._resultset.index_of(str_index)
|
if int_index >= len(self._fields):
|
raise IndexError("Argument out of range")
|
if int_index == -1:
|
raise ValueError(f"Column name '{str_index}' not found")
|
return str(self._fields[int_index])
|
|
|
class BaseResult:
|
"""Provides base functionality for result objects.
|
|
Args:
|
connection (mysqlx.connection.Connection): The Connection object.
|
"""
|
|
def __init__(self, connection: ConnectionType) -> None:
|
self._connection: ConnectionType = connection
|
self._closed: bool = False
|
self._rows_affected: int = 0
|
self._generated_id: int = -1
|
self._generated_ids: List[int] = []
|
self._warnings: List[Dict[str, Union[int, str]]] = []
|
|
if connection is None:
|
self._protocol = None
|
else:
|
self._protocol = connection.protocol
|
connection.fetch_active_result()
|
|
def get_affected_items_count(self) -> int:
|
"""Returns the number of affected items for the last operation.
|
|
Returns:
|
int: The number of affected items.
|
"""
|
return self._rows_affected
|
|
def get_warnings(self) -> List[Dict[str, Union[int, str]]]:
|
"""Returns the warnings.
|
|
Returns:
|
`list`: The list of warnings.
|
"""
|
return self._warnings
|
|
def get_warnings_count(self) -> int:
|
"""Returns the number of warnings.
|
|
Returns:
|
int: The number of warnings.
|
"""
|
return len(self._warnings)
|
|
def set_closed(self, flag: bool) -> None:
|
"""Sets if resultset fetch is done."""
|
self._closed = flag
|
|
def append_warning(self, level: int, code: int, msg: str) -> None:
|
"""Append a warning.
|
|
Args:
|
level (int): The warning level.
|
code (int): The warning code.
|
msg (str): The warning message.
|
"""
|
self._warnings.append({"level": level, "code": code, "msg": msg})
|
|
def set_generated_ids(self, generated_ids: List[int]) -> None:
|
"""Sets the generated ids."""
|
self._generated_ids = generated_ids
|
|
def set_generated_insert_id(self, generated_id: int) -> None:
|
"""Sets the generated insert id."""
|
self._generated_id = generated_id
|
|
def set_rows_affected(self, total: int) -> None:
|
"""Sets the number of rows affected."""
|
self._rows_affected = total
|
|
|
class Result(BaseResult):
|
"""Allows retrieving information about non query operations performed on
|
the database.
|
|
Args:
|
connection (mysqlx.connection.Connection): The Connection object.
|
ids (`list`): A list of IDs.
|
"""
|
|
def __init__(
|
self,
|
connection: Optional[ConnectionType] = None,
|
ids: Optional[List[int]] = None,
|
) -> None:
|
super().__init__(connection)
|
self._ids: Optional[List[int]] = ids
|
|
if connection is not None:
|
self._connection.close_result(self)
|
|
def get_autoincrement_value(self) -> int:
|
"""Returns the last insert id auto generated.
|
|
Returns:
|
int: The last insert id.
|
"""
|
return self._generated_id
|
|
@deprecated("8.0.12")
|
def get_document_id(self) -> Optional[int]:
|
"""Returns ID of the last document inserted into a collection.
|
|
.. deprecated:: 8.0.12
|
"""
|
if self._ids is None or len(self._ids) == 0:
|
return None
|
return self._ids[0]
|
|
@deprecated("8.0.12")
|
def get_generated_insert_id(self) -> int:
|
"""Returns the generated insert id.
|
|
.. deprecated:: 8.0.12
|
"""
|
return self._generated_id
|
|
def get_generated_ids(self) -> List[int]:
|
"""Returns the generated ids."""
|
return self._generated_ids
|
|
|
class BufferingResult(BaseResult):
|
"""Provides base functionality for buffering result objects.
|
|
Args:
|
connection (mysqlx.connection.Connection): The Connection object.
|
ids (`list`): A list of IDs.
|
"""
|
|
def __init__(self, connection: ConnectionType) -> None:
|
super().__init__(connection)
|
self._columns: List[Column] = []
|
self._has_data: bool = False
|
self._has_more_results: bool = False
|
self._items: List[Union[Row, DbDoc]] = []
|
self._page_size: int = 0
|
self._position: int = -1
|
self._init_result()
|
|
def __getitem__(self, index: int) -> Union[Row, DbDoc]:
|
return self._items[index]
|
|
@property
|
def count(self) -> int:
|
"""int: The total of items."""
|
return len(self._items)
|
|
def _init_result(self) -> None:
|
"""Initialize the result."""
|
self._columns = self._connection.get_column_metadata(self)
|
self._has_more_data = len(self._columns) > 0
|
self._items = []
|
self._page_size = 20
|
self._position = -1
|
self._connection.set_active_result(self if self._has_more_data else None)
|
|
def _read_item(self, dumping: bool) -> Optional[Union[Row, DbDoc]]:
|
"""Read item.
|
|
Args:
|
dumping (bool): `True` for dumping.
|
|
Returns:
|
:class:`mysqlx.Row`: A `Row` object.
|
"""
|
row = self._connection.read_row(self)
|
if row is None:
|
return None
|
item = [None] * len(row["field"])
|
if not dumping:
|
for key in range(len(row["field"])):
|
column = self._columns[key]
|
item[key] = from_protobuf(column, row["field"][key])
|
return Row(self, item)
|
|
def _page_in_items(self) -> Union[bool, int]:
|
"""Reads the page items.
|
|
Returns:
|
int: Total items read.
|
"""
|
if self._closed:
|
return False
|
|
count = 0
|
for _ in range(self._page_size):
|
item = self._read_item(False)
|
if item is None:
|
break
|
self._items.append(item)
|
count += 1
|
return count
|
|
def index_of(self, col_name: str) -> int:
|
"""Returns the index of the column.
|
|
Returns:
|
int: The index of the column.
|
"""
|
index = 0
|
for col in self._columns:
|
if col.get_column_label() == col_name:
|
return index
|
index += 1
|
return -1
|
|
def fetch_one(self) -> Optional[Union[Row, DbDoc]]:
|
"""Fetch one item.
|
|
Returns:
|
:class:`mysqlx.Row` or :class:`mysqlx.DbDoc`: one result item.
|
"""
|
if self._closed:
|
return None
|
|
return self._read_item(False)
|
|
def fetch_all(self) -> List[Union[Row, DbDoc]]:
|
"""Fetch all items.
|
|
Returns:
|
`list`: The list of items of :class:`mysqlx.DbDoc` or
|
:class:`mysqlx.Row`.
|
"""
|
while True:
|
if not self._page_in_items():
|
break
|
return self._items
|
|
def set_has_data(self, flag: bool) -> None:
|
"""Sets if result has data.
|
|
Args:
|
flag (bool): `True` if result has data.
|
"""
|
self._has_data = flag
|
|
def set_has_more_results(self, flag: bool) -> None:
|
"""Sets if has more results.
|
|
Args:
|
flag (bool): `True` if has more results.
|
"""
|
self._has_more_results = flag
|
|
|
class RowResult(BufferingResult):
|
"""Allows traversing the Row objects returned by a Table.select operation.
|
|
Args:
|
connection (mysqlx.connection.Connection): The Connection object.
|
"""
|
|
@property
|
def columns(self) -> List[Column]:
|
"""`list`: The list of columns."""
|
return self._columns
|
|
def get_columns(self) -> List[Column]:
|
"""Returns the list of columns.
|
|
Returns:
|
`list`: The list of columns.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._columns
|
|
|
class SqlResult(RowResult):
|
"""Represents a result from a SQL statement.
|
|
Args:
|
connection (mysqlx.connection.Connection): The Connection object.
|
"""
|
|
def get_autoincrement_value(self) -> int:
|
"""Returns the identifier for the last record inserted.
|
|
Returns:
|
str: The identifier of the last record inserted.
|
"""
|
return self._generated_id
|
|
def next_result(self) -> bool:
|
"""Process the next result.
|
|
Returns:
|
bool: Returns `True` if the fetch is done.
|
"""
|
if self._closed:
|
return False
|
self._has_more_results = False
|
self._init_result()
|
return True
|
|
def has_data(self) -> bool:
|
"""Returns True if result has data.
|
|
Returns:
|
bool: Returns `True` if result has data.
|
|
.. versionadded:: 8.0.12
|
"""
|
return self._has_data
|
|
|
class DocResult(BufferingResult):
|
"""Allows traversing the DbDoc objects returned by a Collection.find
|
operation.
|
|
Args:
|
connection (mysqlx.connection.Connection): The Connection object.
|
"""
|
|
def _read_item(self, dumping: bool) -> DbDoc:
|
"""Read item.
|
|
Args:
|
dumping (bool): `True` for dumping.
|
|
Returns:
|
:class:`mysqlx.DbDoc`: A `DbDoc` object.
|
"""
|
row = super()._read_item(dumping)
|
if row is None:
|
return None
|
return DbDoc(decode_from_bytes(row[0])) # type: ignore[index]
|
