# 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 X protocol for MySQL servers."""
|
|
import struct
|
import zlib
|
|
from io import BytesIO
|
from typing import Any, Dict, List, Optional, Tuple, Union
|
|
try:
|
import lz4.frame
|
|
HAVE_LZ4 = True
|
except ImportError:
|
HAVE_LZ4 = False
|
|
try:
|
import zstandard as zstd
|
|
HAVE_ZSTD = True
|
except ImportError:
|
HAVE_ZSTD = False
|
|
from .errors import (
|
InterfaceError,
|
NotSupportedError,
|
OperationalError,
|
ProgrammingError,
|
)
|
from .expr import (
|
ExprParser,
|
build_bool_scalar,
|
build_expr,
|
build_int_scalar,
|
build_scalar,
|
build_unsigned_int_scalar,
|
)
|
from .helpers import encode_to_bytes, get_item_or_attr
|
from .logger import logger
|
from .protobuf import (
|
CRUD_PREPARE_MAPPING,
|
PROTOBUF_REPEATED_TYPES,
|
SERVER_MESSAGES,
|
Message,
|
mysqlxpb_enum,
|
)
|
from .result import Column
|
from .statement import (
|
AddStatement,
|
DeleteStatement,
|
FilterableStatement,
|
FindStatement,
|
InsertStatement,
|
ModifyStatement,
|
ReadStatement,
|
RemoveStatement,
|
SqlStatement,
|
UpdateStatement,
|
)
|
from .types import (
|
ColumnType,
|
MessageType,
|
ProtobufMessageCextType,
|
ProtobufMessageType,
|
ResultBaseType,
|
SocketType,
|
StatementType,
|
StrOrBytes,
|
)
|
|
_COMPRESSION_THRESHOLD = 1000
|
|
|
class Compressor:
|
"""Implements compression/decompression using `zstd_stream`, `lz4_message`
|
and `deflate_stream` algorithms.
|
|
Args:
|
algorithm (str): Compression algorithm.
|
|
.. versionadded:: 8.0.21
|
|
"""
|
|
def __init__(self, algorithm: str) -> None:
|
self._algorithm: str = algorithm
|
self._compressobj: Any = None
|
self._decompressobj: Any = None
|
|
if algorithm == "zstd_stream":
|
self._compressobj = zstd.ZstdCompressor()
|
self._decompressobj = zstd.ZstdDecompressor()
|
elif algorithm == "deflate_stream":
|
self._compressobj = zlib.compressobj()
|
self._decompressobj = zlib.decompressobj()
|
|
def compress(self, data: StrOrBytes) -> bytes:
|
"""Compresses data and returns it.
|
|
Args:
|
data (str, bytes or buffer object): Data to be compressed.
|
|
Returns:
|
bytes: Compressed data.
|
"""
|
if self._algorithm == "zstd_stream":
|
return self._compressobj.compress(data)
|
if self._algorithm == "lz4_message":
|
with lz4.frame.LZ4FrameCompressor() as compressor:
|
compressed = compressor.begin()
|
compressed += compressor.compress(data)
|
compressed += compressor.flush()
|
return compressed
|
|
# Using 'deflate_stream' algorithm
|
compressed = self._compressobj.compress(data)
|
compressed += self._compressobj.flush(zlib.Z_SYNC_FLUSH)
|
return compressed
|
|
def decompress(self, data: StrOrBytes) -> bytes:
|
"""Decompresses a frame of data and returns it as a string of bytes.
|
|
Args:
|
data (str, bytes or buffer object): Data to be compressed.
|
|
Returns:
|
bytes: Decompresssed data.
|
"""
|
if self._algorithm == "zstd_stream":
|
return self._decompressobj.decompress(data)
|
if self._algorithm == "lz4_message":
|
with lz4.frame.LZ4FrameDecompressor() as decompressor:
|
decompressed = decompressor.decompress(data)
|
return decompressed
|
|
# Using 'deflate' algorithm
|
decompressed = self._decompressobj.decompress(data)
|
decompressed += self._decompressobj.flush(zlib.Z_SYNC_FLUSH)
|
return decompressed
|
|
|
class MessageReader:
|
"""Implements a Message Reader.
|
|
Args:
|
socket_stream (mysqlx.connection.SocketStream): `SocketStream` object.
|
|
.. versionadded:: 8.0.21
|
"""
|
|
def __init__(self, socket_stream: SocketType) -> None:
|
self._stream: SocketType = socket_stream
|
self._compressor: Optional[Compressor] = None
|
self._msg: MessageType = None
|
self._msg_queue: List[Message] = []
|
|
def _read_message(self) -> MessageType:
|
"""Reads X Protocol messages from the stream and returns a
|
:class:`mysqlx.protobuf.Message` object.
|
|
Raises:
|
:class:`mysqlx.ProgrammingError`: If e connected server does not
|
have the MySQL X protocol plugin
|
enabled.
|
|
Returns:
|
mysqlx.protobuf.Message: MySQL X Protobuf Message.
|
"""
|
if self._msg_queue:
|
return self._msg_queue.pop(0)
|
|
frame_size, frame_type = struct.unpack("<LB", self._stream.read(5))
|
|
if frame_type == 10:
|
raise ProgrammingError(
|
"The connected server does not have the "
|
"MySQL X protocol plugin enabled or "
|
"protocol mismatch"
|
)
|
|
frame_payload = self._stream.read(frame_size - 1)
|
if frame_type not in SERVER_MESSAGES:
|
raise ValueError(f"Unknown message type: {frame_type}")
|
|
# Do not parse empty notices, Message requires a type in payload
|
if frame_type == 11 and frame_payload == b"":
|
return self._read_message()
|
|
frame_msg = Message.from_server_message(frame_type, frame_payload)
|
|
if frame_type == 19: # Mysqlx.ServerMessages.Type.COMPRESSION
|
uncompressed_size = frame_msg["uncompressed_size"]
|
stream = BytesIO(self._compressor.decompress(frame_msg["payload"]))
|
bytes_processed = 0
|
while bytes_processed < uncompressed_size:
|
payload_size, msg_type = struct.unpack("<LB", stream.read(5))
|
payload = stream.read(payload_size - 1)
|
self._msg_queue.append(Message.from_server_message(msg_type, payload))
|
bytes_processed += payload_size + 4
|
return self._msg_queue.pop(0) if self._msg_queue else None
|
|
return frame_msg
|
|
def read_message(self) -> MessageType:
|
"""Read message.
|
|
Returns:
|
mysqlx.protobuf.Message: MySQL X Protobuf Message.
|
"""
|
if self._msg is not None:
|
msg = self._msg
|
self._msg = None
|
return msg
|
return self._read_message()
|
|
def push_message(self, msg: MessageType) -> None:
|
"""Push message.
|
|
Args:
|
msg (mysqlx.protobuf.Message): MySQL X Protobuf Message.
|
|
Raises:
|
:class:`mysqlx.OperationalError`: If message push slot is full.
|
"""
|
if self._msg is not None:
|
raise OperationalError("Message push slot is full")
|
self._msg = msg
|
|
def set_compression(self, algorithm: str) -> None:
|
"""Creates a :class:`mysqlx.protocol.Compressor` object based on the
|
compression algorithm.
|
|
Args:
|
algorithm (str): Compression algorithm.
|
|
.. versionadded:: 8.0.21
|
|
"""
|
self._compressor = Compressor(algorithm) if algorithm else None
|
|
|
class MessageWriter:
|
"""Implements a Message Writer.
|
|
Args:
|
socket_stream (mysqlx.connection.SocketStream): `SocketStream` object.
|
|
.. versionadded:: 8.0.21
|
|
"""
|
|
def __init__(self, socket_stream: SocketType) -> None:
|
self._stream: SocketType = socket_stream
|
self._compressor: Optional[Compressor] = None
|
|
def write_message(self, msg_type: int, msg: MessageType) -> None:
|
"""Write message.
|
|
Args:
|
msg_type (int): The message type.
|
msg (mysqlx.protobuf.Message): MySQL X Protobuf Message.
|
"""
|
msg_size = msg.byte_size(msg)
|
if self._compressor and msg_size > _COMPRESSION_THRESHOLD:
|
msg_str = encode_to_bytes(msg.serialize_to_string())
|
header = struct.pack("<LB", msg_size + 1, msg_type)
|
compressed = self._compressor.compress(b"".join([header, msg_str]))
|
|
msg_first_fields = Message("Mysqlx.Connection.Compression")
|
msg_first_fields["client_messages"] = msg_type
|
msg_first_fields["uncompressed_size"] = msg_size + 5
|
|
msg_payload = Message("Mysqlx.Connection.Compression")
|
msg_payload["payload"] = compressed
|
|
output = b"".join(
|
[
|
encode_to_bytes(msg_first_fields.serialize_partial_to_string())[
|
:-2
|
],
|
encode_to_bytes(msg_payload.serialize_partial_to_string()),
|
]
|
)
|
|
msg_comp_id = mysqlxpb_enum("Mysqlx.ClientMessages.Type.COMPRESSION")
|
header = struct.pack("<LB", len(output) + 1, msg_comp_id)
|
self._stream.sendall(b"".join([header, output]))
|
else:
|
msg_str = encode_to_bytes(msg.serialize_to_string())
|
header = struct.pack("<LB", msg_size + 1, msg_type)
|
self._stream.sendall(b"".join([header, msg_str]))
|
|
def set_compression(self, algorithm: str) -> None:
|
"""Creates a :class:`mysqlx.protocol.Compressor` object based on the
|
compression algorithm.
|
|
Args:
|
algorithm (str): Compression algorithm.
|
"""
|
self._compressor = Compressor(algorithm) if algorithm else None
|
|
|
class Protocol:
|
"""Implements the MySQL X Protocol.
|
|
Args:
|
read (mysqlx.protocol.MessageReader): A Message Reader object.
|
writer (mysqlx.protocol.MessageWriter): A Message Writer object.
|
|
.. versionchanged:: 8.0.21
|
"""
|
|
def __init__(self, reader: MessageReader, writer: MessageWriter) -> None:
|
self._reader: MessageReader = reader
|
self._writer: MessageWriter = writer
|
self._compression_algorithm: Optional[str] = None
|
self._warnings: List[str] = []
|
|
@property
|
def compression_algorithm(self) -> Optional[str]:
|
"""str: The compresion algorithm."""
|
return self._compression_algorithm
|
|
@staticmethod
|
def _apply_filter(msg: MessageType, stmt: FilterableStatement) -> None:
|
"""Apply filter.
|
|
Args:
|
msg (mysqlx.protobuf.Message): The MySQL X Protobuf Message.
|
stmt (Statement): A `Statement` based type object.
|
"""
|
if stmt.has_where:
|
msg["criteria"] = stmt.get_where_expr()
|
if stmt.has_sort:
|
msg["order"].extend(stmt.get_sort_expr())
|
if stmt.has_group_by:
|
msg["grouping"].extend(stmt.get_grouping())
|
if stmt.has_having:
|
msg["grouping_criteria"] = stmt.get_having()
|
|
def _create_any(self, arg: Any) -> Optional[MessageType]:
|
"""Create any.
|
|
Args:
|
arg (object): Arbitrary object.
|
|
Returns:
|
mysqlx.protobuf.Message: MySQL X Protobuf Message.
|
"""
|
if isinstance(arg, str):
|
value = Message("Mysqlx.Datatypes.Scalar.String", value=arg)
|
scalar = Message("Mysqlx.Datatypes.Scalar", type=8, v_string=value)
|
return Message("Mysqlx.Datatypes.Any", type=1, scalar=scalar)
|
if isinstance(arg, bool):
|
return Message(
|
"Mysqlx.Datatypes.Any", type=1, scalar=build_bool_scalar(arg)
|
)
|
if isinstance(arg, int):
|
if arg < 0:
|
return Message(
|
"Mysqlx.Datatypes.Any",
|
type=1,
|
scalar=build_int_scalar(arg),
|
)
|
return Message(
|
"Mysqlx.Datatypes.Any",
|
type=1,
|
scalar=build_unsigned_int_scalar(arg),
|
)
|
if isinstance(arg, tuple) and len(arg) == 2:
|
arg_key, arg_value = arg
|
obj_fld = Message(
|
"Mysqlx.Datatypes.Object.ObjectField",
|
key=arg_key,
|
value=self._create_any(arg_value),
|
)
|
obj = Message("Mysqlx.Datatypes.Object", fld=[obj_fld.get_message()])
|
return Message("Mysqlx.Datatypes.Any", type=2, obj=obj)
|
if isinstance(arg, dict) or (
|
isinstance(arg, (list, tuple)) and isinstance(arg[0], dict)
|
):
|
array_values = []
|
for items in arg:
|
obj_flds = []
|
for key, value in items.items():
|
# Array can only handle Any types, Mysqlx.Datatypes.Any.obj
|
obj_fld = Message(
|
"Mysqlx.Datatypes.Object.ObjectField",
|
key=key,
|
value=self._create_any(value),
|
)
|
obj_flds.append(obj_fld.get_message())
|
msg_obj = Message("Mysqlx.Datatypes.Object", fld=obj_flds)
|
msg_any = Message("Mysqlx.Datatypes.Any", type=2, obj=msg_obj)
|
array_values.append(msg_any.get_message())
|
|
msg = Message("Mysqlx.Datatypes.Array")
|
msg["value"] = array_values
|
return Message("Mysqlx.Datatypes.Any", type=3, array=msg)
|
if isinstance(arg, list):
|
obj_flds = []
|
for key, value in arg:
|
obj_fld = Message(
|
"Mysqlx.Datatypes.Object.ObjectField",
|
key=key,
|
value=self._create_any(value),
|
)
|
obj_flds.append(obj_fld.get_message())
|
msg_obj = Message("Mysqlx.Datatypes.Object", fld=obj_flds)
|
msg_any = Message("Mysqlx.Datatypes.Any", type=2, obj=msg_obj)
|
return msg_any
|
|
return None
|
|
def _get_binding_args(
|
self, stmt: Union[FilterableStatement, SqlStatement], is_scalar: bool = True
|
) -> Union[List[None], List[Union[ProtobufMessageType, ProtobufMessageCextType]]]:
|
"""Returns the binding any/scalar.
|
|
Args:
|
stmt (Statement): A `Statement` based type object.
|
is_scalar (bool): `True` to return scalar values.
|
|
Raises:
|
:class:`mysqlx.ProgrammingError`: If unable to find placeholder for
|
parameter.
|
|
Returns:
|
list: A list of ``Any`` or ``Scalar`` objects.
|
"""
|
|
def build_value(
|
value: Any,
|
) -> Union[ProtobufMessageType, ProtobufMessageCextType]:
|
if is_scalar:
|
return build_scalar(value).get_message()
|
return self._create_any(value).get_message()
|
|
bindings = stmt.get_bindings()
|
binding_map = stmt.get_binding_map()
|
|
# If binding_map is None it's a SqlStatement object
|
if binding_map is None:
|
return [build_value(value) for value in bindings]
|
|
count = len(binding_map)
|
args: List[Any] = count * [None]
|
if count != len(bindings):
|
raise ProgrammingError(
|
"The number of bind parameters and placeholders do not match"
|
)
|
for name, value in bindings.items(): # type: ignore[union-attr]
|
if name not in binding_map:
|
raise ProgrammingError(
|
f"Unable to find placeholder for parameter: {name}"
|
)
|
pos = binding_map[name]
|
args[pos] = build_value(value)
|
return args
|
|
def _process_frame(self, msg: MessageType, result: ResultBaseType) -> None:
|
"""Process frame.
|
|
Args:
|
msg (mysqlx.protobuf.Message): A MySQL X Protobuf Message.
|
result (Result): A `Result` based type object.
|
"""
|
if msg["type"] == 1:
|
warn_msg = Message.from_message("Mysqlx.Notice.Warning", msg["payload"])
|
self._warnings.append(warn_msg.msg)
|
logger.warning(
|
"Protocol.process_frame Received Warning Notice code %s: %s",
|
warn_msg.code,
|
warn_msg.msg,
|
)
|
result.append_warning(warn_msg.level, warn_msg.code, warn_msg.msg)
|
elif msg["type"] == 2:
|
Message.from_message("Mysqlx.Notice.SessionVariableChanged", msg["payload"])
|
elif msg["type"] == 3:
|
sess_state_msg = Message.from_message(
|
"Mysqlx.Notice.SessionStateChanged", msg["payload"]
|
)
|
if sess_state_msg["param"] == mysqlxpb_enum(
|
"Mysqlx.Notice.SessionStateChanged.Parameter.GENERATED_DOCUMENT_IDS"
|
):
|
result.set_generated_ids(
|
[
|
get_item_or_attr(
|
get_item_or_attr(value, "v_octets"), "value"
|
).decode()
|
for value in sess_state_msg["value"]
|
]
|
)
|
else: # Following results are unitary and not a list
|
sess_state_value = (
|
sess_state_msg["value"][0]
|
if isinstance(
|
sess_state_msg["value"], tuple(PROTOBUF_REPEATED_TYPES)
|
)
|
else sess_state_msg["value"]
|
)
|
if sess_state_msg["param"] == mysqlxpb_enum(
|
"Mysqlx.Notice.SessionStateChanged.Parameter.ROWS_AFFECTED"
|
):
|
result.set_rows_affected(
|
get_item_or_attr(sess_state_value, "v_unsigned_int")
|
)
|
elif sess_state_msg["param"] == mysqlxpb_enum(
|
"Mysqlx.Notice.SessionStateChanged.Parameter.GENERATED_INSERT_ID"
|
):
|
result.set_generated_insert_id(
|
get_item_or_attr(sess_state_value, "v_unsigned_int")
|
)
|
|
def _read_message(self, result: ResultBaseType) -> Optional[MessageType]:
|
"""Read message.
|
|
Args:
|
result (Result): A `Result` based type object.
|
"""
|
while True:
|
try:
|
msg = self._reader.read_message()
|
except RuntimeError as err:
|
warnings = repr(result.get_warnings())
|
if warnings:
|
raise RuntimeError(f"{err} reason: {warnings}") from err
|
if msg.type == "Mysqlx.Error":
|
raise OperationalError(msg["msg"], msg["code"])
|
if msg.type == "Mysqlx.Notice.Frame":
|
try:
|
self._process_frame(msg, result)
|
except (AttributeError, KeyError):
|
continue
|
elif msg.type == "Mysqlx.Sql.StmtExecuteOk":
|
return None
|
elif msg.type == "Mysqlx.Resultset.FetchDone":
|
result.set_closed(True)
|
elif msg.type == "Mysqlx.Resultset.FetchDoneMoreResultsets":
|
result.set_has_more_results(True)
|
elif msg.type == "Mysqlx.Resultset.Row":
|
result.set_has_data(True)
|
break
|
else:
|
break
|
return msg
|
|
def set_compression(self, algorithm: str) -> None:
|
"""Sets the compression algorithm to be used by the compression
|
object, for uplink and downlink.
|
|
Args:
|
algorithm (str): Algorithm to be used in compression/decompression.
|
|
.. versionadded:: 8.0.21
|
|
"""
|
self._compression_algorithm = algorithm
|
self._reader.set_compression(algorithm)
|
self._writer.set_compression(algorithm)
|
|
def get_capabilites(self) -> MessageType:
|
"""Get capabilities.
|
|
Returns:
|
mysqlx.protobuf.Message: MySQL X Protobuf Message.
|
"""
|
msg = Message("Mysqlx.Connection.CapabilitiesGet")
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.CON_CAPABILITIES_GET"),
|
msg,
|
)
|
msg = self._reader.read_message()
|
while msg.type == "Mysqlx.Notice.Frame":
|
msg = self._reader.read_message()
|
|
if msg.type == "Mysqlx.Error":
|
raise OperationalError(msg["msg"], msg["code"])
|
|
return msg
|
|
def set_capabilities(self, **kwargs: Any) -> None:
|
"""Set capabilities.
|
|
Args:
|
**kwargs: Arbitrary keyword arguments.
|
|
Returns:
|
mysqlx.protobuf.Message: MySQL X Protobuf Message.
|
"""
|
if not kwargs:
|
return None
|
capabilities = Message("Mysqlx.Connection.Capabilities")
|
for key, value in kwargs.items():
|
capability = Message("Mysqlx.Connection.Capability")
|
capability["name"] = key
|
if isinstance(value, dict):
|
items = value
|
obj_flds = []
|
for item in items:
|
obj_fld = Message(
|
"Mysqlx.Datatypes.Object.ObjectField",
|
key=item,
|
value=self._create_any(items[item]),
|
)
|
obj_flds.append(obj_fld.get_message())
|
msg_obj = Message("Mysqlx.Datatypes.Object", fld=obj_flds)
|
msg_any = Message("Mysqlx.Datatypes.Any", type=2, obj=msg_obj)
|
capability["value"] = msg_any.get_message()
|
else:
|
capability["value"] = self._create_any(value)
|
|
capabilities["capabilities"].extend([capability.get_message()])
|
msg = Message("Mysqlx.Connection.CapabilitiesSet")
|
msg["capabilities"] = capabilities
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.CON_CAPABILITIES_SET"),
|
msg,
|
)
|
|
try:
|
return self.read_ok()
|
except InterfaceError as err:
|
# Skip capability "session_connect_attrs" error since
|
# is only available on version >= 8.0.16
|
if err.errno != 5002:
|
raise
|
return None
|
|
def send_auth_start(
|
self,
|
method: str,
|
auth_data: Optional[str] = None,
|
initial_response: Optional[str] = None,
|
) -> None:
|
"""Send authenticate start.
|
|
Args:
|
method (str): Message method.
|
auth_data (Optional[str]): Authentication data.
|
initial_response (Optional[str]): Initial response.
|
"""
|
msg = Message("Mysqlx.Session.AuthenticateStart")
|
msg["mech_name"] = method
|
if auth_data is not None:
|
msg["auth_data"] = auth_data
|
if initial_response is not None:
|
msg["initial_response"] = initial_response
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.SESS_AUTHENTICATE_START"),
|
msg,
|
)
|
|
def read_auth_continue(self) -> bytes:
|
"""Read authenticate continue.
|
|
Raises:
|
:class:`InterfaceError`: If the message type is not
|
`Mysqlx.Session.AuthenticateContinue`
|
|
Returns:
|
str: The authentication data.
|
"""
|
msg = self._reader.read_message()
|
while msg.type == "Mysqlx.Notice.Frame":
|
msg = self._reader.read_message()
|
if msg.type != "Mysqlx.Session.AuthenticateContinue":
|
raise InterfaceError(
|
"Unexpected message encountered during authentication handshake"
|
)
|
return msg["auth_data"]
|
|
def send_auth_continue(self, auth_data: str) -> None:
|
"""Send authenticate continue.
|
|
Args:
|
auth_data (str): Authentication data.
|
"""
|
msg = Message("Mysqlx.Session.AuthenticateContinue", auth_data=auth_data)
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.SESS_AUTHENTICATE_CONTINUE"),
|
msg,
|
)
|
|
def read_auth_ok(self) -> None:
|
"""Read authenticate OK.
|
|
Raises:
|
:class:`mysqlx.InterfaceError`: If message type is `Mysqlx.Error`.
|
"""
|
while True:
|
msg = self._reader.read_message()
|
if msg.type == "Mysqlx.Session.AuthenticateOk":
|
break
|
if msg.type == "Mysqlx.Error":
|
raise InterfaceError(msg.msg)
|
|
def send_prepare_prepare(
|
self,
|
msg_type: str,
|
msg: MessageType,
|
stmt: Union[
|
FindStatement,
|
DeleteStatement,
|
ModifyStatement,
|
ReadStatement,
|
RemoveStatement,
|
UpdateStatement,
|
],
|
) -> None:
|
"""
|
Send prepare statement.
|
|
Args:
|
msg_type (str): Message ID string.
|
msg (mysqlx.protobuf.Message): MySQL X Protobuf Message.
|
stmt (Statement): A `Statement` based type object.
|
|
Raises:
|
:class:`mysqlx.NotSupportedError`: If prepared statements are not
|
supported.
|
|
.. versionadded:: 8.0.16
|
"""
|
if stmt.has_limit and msg.type != "Mysqlx.Crud.Insert":
|
# Remove 'limit' from message by building a new one
|
if msg.type == "Mysqlx.Crud.Find":
|
_, msg = self.build_find(stmt) # type: ignore[arg-type]
|
elif msg.type == "Mysqlx.Crud.Update":
|
_, msg = self.build_update(stmt) # type: ignore[arg-type]
|
elif msg.type == "Mysqlx.Crud.Delete":
|
_, msg = self.build_delete(stmt) # type: ignore[arg-type]
|
else:
|
raise ValueError(f"Invalid message type: {msg_type}")
|
# Build 'limit_expr' message
|
position = len(stmt.get_bindings())
|
placeholder = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.PLACEHOLDER")
|
msg_limit_expr = Message("Mysqlx.Crud.LimitExpr")
|
msg_limit_expr["row_count"] = Message(
|
"Mysqlx.Expr.Expr", type=placeholder, position=position
|
)
|
if msg.type == "Mysqlx.Crud.Find":
|
msg_limit_expr["offset"] = Message(
|
"Mysqlx.Expr.Expr", type=placeholder, position=position + 1
|
)
|
msg["limit_expr"] = msg_limit_expr
|
|
oneof_type, oneof_op = CRUD_PREPARE_MAPPING[msg_type]
|
msg_oneof = Message("Mysqlx.Prepare.Prepare.OneOfMessage")
|
msg_oneof["type"] = mysqlxpb_enum(oneof_type)
|
msg_oneof[oneof_op] = msg
|
msg_prepare = Message("Mysqlx.Prepare.Prepare")
|
msg_prepare["stmt_id"] = stmt.stmt_id
|
msg_prepare["stmt"] = msg_oneof
|
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.PREPARE_PREPARE"),
|
msg_prepare,
|
)
|
|
try:
|
self.read_ok()
|
except InterfaceError as err:
|
raise NotSupportedError from err
|
|
def send_prepare_execute(
|
self, msg_type: str, msg: MessageType, stmt: FilterableStatement
|
) -> None:
|
"""
|
Send execute statement.
|
|
Args:
|
msg_type (str): Message ID string.
|
msg (mysqlx.protobuf.Message): MySQL X Protobuf Message.
|
stmt (Statement): A `Statement` based type object.
|
|
.. versionadded:: 8.0.16
|
"""
|
oneof_type, oneof_op = CRUD_PREPARE_MAPPING[msg_type]
|
msg_oneof = Message("Mysqlx.Prepare.Prepare.OneOfMessage")
|
msg_oneof["type"] = mysqlxpb_enum(oneof_type)
|
msg_oneof[oneof_op] = msg
|
msg_execute = Message("Mysqlx.Prepare.Execute")
|
msg_execute["stmt_id"] = stmt.stmt_id
|
|
args = self._get_binding_args(stmt, is_scalar=False)
|
if args:
|
msg_execute["args"].extend(args)
|
|
if stmt.has_limit:
|
msg_execute["args"].extend(
|
[
|
self._create_any(stmt.get_limit_row_count()).get_message(),
|
self._create_any(stmt.get_limit_offset()).get_message(),
|
]
|
)
|
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.PREPARE_EXECUTE"),
|
msg_execute,
|
)
|
|
def send_prepare_deallocate(self, stmt_id: int) -> None:
|
"""
|
Send prepare deallocate statement.
|
|
Args:
|
stmt_id (int): Statement ID.
|
|
.. versionadded:: 8.0.16
|
"""
|
msg_dealloc = Message("Mysqlx.Prepare.Deallocate")
|
msg_dealloc["stmt_id"] = stmt_id
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.PREPARE_DEALLOCATE"),
|
msg_dealloc,
|
)
|
self.read_ok()
|
|
def send_msg_without_ps(
|
self,
|
msg_type: str,
|
msg: MessageType,
|
stmt: Union[FilterableStatement, SqlStatement],
|
) -> None:
|
"""
|
Send a message without prepared statements support.
|
|
Args:
|
msg_type (str): Message ID string.
|
msg (mysqlx.protobuf.Message): MySQL X Protobuf Message.
|
stmt (Statement): A `Statement` based type object.
|
|
.. versionadded:: 8.0.16
|
"""
|
if stmt.has_limit:
|
msg_limit = Message("Mysqlx.Crud.Limit")
|
msg_limit["row_count"] = stmt.get_limit_row_count() # type: ignore[union-attr]
|
if msg.type == "Mysqlx.Crud.Find":
|
msg_limit["offset"] = stmt.get_limit_offset() # type: ignore[union-attr]
|
msg["limit"] = msg_limit
|
is_scalar = msg_type != "Mysqlx.ClientMessages.Type.SQL_STMT_EXECUTE"
|
args = self._get_binding_args(stmt, is_scalar=is_scalar)
|
if args:
|
msg["args"].extend(args)
|
self.send_msg(msg_type, msg)
|
|
def send_msg(self, msg_type: str, msg: MessageType) -> None:
|
"""
|
Send a message.
|
|
Args:
|
msg_type (str): Message ID string.
|
msg (mysqlx.protobuf.Message): MySQL X Protobuf Message.
|
|
.. versionadded:: 8.0.16
|
"""
|
self._writer.write_message(mysqlxpb_enum(msg_type), msg)
|
|
def build_find(
|
self, stmt: Union[FindStatement, ReadStatement]
|
) -> Tuple[str, MessageType]:
|
"""Build find/read message.
|
|
Args:
|
stmt (Statement): A :class:`mysqlx.ReadStatement` or
|
:class:`mysqlx.FindStatement` object.
|
|
Returns:
|
(tuple): Tuple containing:
|
|
* `str`: Message ID string.
|
* :class:`mysqlx.protobuf.Message`: MySQL X Protobuf Message.
|
|
.. versionadded:: 8.0.16
|
"""
|
data_model = mysqlxpb_enum(
|
"Mysqlx.Crud.DataModel.DOCUMENT"
|
if stmt.is_doc_based()
|
else "Mysqlx.Crud.DataModel.TABLE"
|
)
|
collection = Message(
|
"Mysqlx.Crud.Collection",
|
name=stmt.target.name,
|
schema=stmt.schema.name,
|
)
|
msg = Message("Mysqlx.Crud.Find", data_model=data_model, collection=collection)
|
if stmt.has_projection:
|
msg["projection"] = stmt.get_projection_expr()
|
self._apply_filter(msg, stmt)
|
|
if stmt.is_lock_exclusive():
|
msg["locking"] = mysqlxpb_enum("Mysqlx.Crud.Find.RowLock.EXCLUSIVE_LOCK")
|
elif stmt.is_lock_shared():
|
msg["locking"] = mysqlxpb_enum("Mysqlx.Crud.Find.RowLock.SHARED_LOCK")
|
|
if stmt.lock_contention.value > 0:
|
msg["locking_options"] = stmt.lock_contention.value
|
|
return "Mysqlx.ClientMessages.Type.CRUD_FIND", msg
|
|
def build_update(
|
self, stmt: Union[ModifyStatement, UpdateStatement]
|
) -> Tuple[str, MessageType]:
|
"""Build update message.
|
|
Args:
|
stmt (Statement): A :class:`mysqlx.ModifyStatement` or
|
:class:`mysqlx.UpdateStatement` object.
|
|
Returns:
|
(tuple): Tuple containing:
|
|
* `str`: Message ID string.
|
* :class:`mysqlx.protobuf.Message`: MySQL X Protobuf Message.
|
|
.. versionadded:: 8.0.16
|
"""
|
data_model = mysqlxpb_enum(
|
"Mysqlx.Crud.DataModel.DOCUMENT"
|
if stmt.is_doc_based()
|
else "Mysqlx.Crud.DataModel.TABLE"
|
)
|
collection = Message(
|
"Mysqlx.Crud.Collection",
|
name=stmt.target.name,
|
schema=stmt.schema.name,
|
)
|
msg = Message(
|
"Mysqlx.Crud.Update", data_model=data_model, collection=collection
|
)
|
self._apply_filter(msg, stmt)
|
for _, update_op in stmt.get_update_ops().items():
|
operation = Message("Mysqlx.Crud.UpdateOperation")
|
operation["operation"] = update_op.update_type
|
operation["source"] = update_op.source
|
if update_op.value is not None:
|
operation["value"] = build_expr(update_op.value)
|
msg["operation"].extend([operation.get_message()])
|
|
return "Mysqlx.ClientMessages.Type.CRUD_UPDATE", msg
|
|
def build_delete(
|
self, stmt: Union[DeleteStatement, RemoveStatement]
|
) -> Tuple[str, MessageType]:
|
"""Build delete message.
|
|
Args:
|
stmt (Statement): A :class:`mysqlx.DeleteStatement` or
|
:class:`mysqlx.RemoveStatement` object.
|
|
Returns:
|
(tuple): Tuple containing:
|
|
* `str`: Message ID string.
|
* :class:`mysqlx.protobuf.Message`: MySQL X Protobuf Message.
|
|
.. versionadded:: 8.0.16
|
"""
|
data_model = mysqlxpb_enum(
|
"Mysqlx.Crud.DataModel.DOCUMENT"
|
if stmt.is_doc_based()
|
else "Mysqlx.Crud.DataModel.TABLE"
|
)
|
collection = Message(
|
"Mysqlx.Crud.Collection",
|
name=stmt.target.name,
|
schema=stmt.schema.name,
|
)
|
msg = Message(
|
"Mysqlx.Crud.Delete", data_model=data_model, collection=collection
|
)
|
self._apply_filter(msg, stmt)
|
return "Mysqlx.ClientMessages.Type.CRUD_DELETE", msg
|
|
def build_execute_statement(
|
self,
|
namespace: str,
|
stmt: Union[str, StatementType],
|
fields: Optional[Dict[str, Any]] = None,
|
) -> Tuple[str, MessageType]:
|
"""Build execute statement.
|
|
Args:
|
namespace (str): The namespace.
|
stmt (Statement): A `Statement` based type object.
|
fields (Optional[dict]): The message fields.
|
|
Returns:
|
(tuple): Tuple containing:
|
|
* `str`: Message ID string.
|
* :class:`mysqlx.protobuf.Message`: MySQL X Protobuf Message.
|
|
.. versionadded:: 8.0.16
|
"""
|
msg = Message(
|
"Mysqlx.Sql.StmtExecute",
|
namespace=namespace,
|
stmt=stmt,
|
compact_metadata=False,
|
)
|
|
if fields:
|
obj_flds = []
|
for key, value in fields.items():
|
obj_fld = Message(
|
"Mysqlx.Datatypes.Object.ObjectField",
|
key=key,
|
value=self._create_any(value),
|
)
|
obj_flds.append(obj_fld.get_message())
|
msg_obj = Message("Mysqlx.Datatypes.Object", fld=obj_flds)
|
msg_any = Message("Mysqlx.Datatypes.Any", type=2, obj=msg_obj)
|
msg["args"] = [msg_any.get_message()]
|
return "Mysqlx.ClientMessages.Type.SQL_STMT_EXECUTE", msg
|
|
@staticmethod
|
def build_insert(
|
stmt: Union[AddStatement, InsertStatement]
|
) -> Tuple[str, MessageType]:
|
"""Build insert statement.
|
|
Args:
|
stmt (Statement): A :class:`mysqlx.AddStatement` or
|
:class:`mysqlx.InsertStatement` object.
|
|
Returns:
|
(tuple): Tuple containing:
|
|
* `str`: Message ID string.
|
* :class:`mysqlx.protobuf.Message`: MySQL X Protobuf Message.
|
|
.. versionadded:: 8.0.16
|
"""
|
data_model = mysqlxpb_enum(
|
"Mysqlx.Crud.DataModel.DOCUMENT"
|
if stmt.is_doc_based()
|
else "Mysqlx.Crud.DataModel.TABLE"
|
)
|
collection = Message(
|
"Mysqlx.Crud.Collection",
|
name=stmt.target.name,
|
schema=stmt.schema.name,
|
)
|
msg = Message(
|
"Mysqlx.Crud.Insert", data_model=data_model, collection=collection
|
)
|
|
if hasattr(stmt, "_fields"):
|
for field in stmt._fields:
|
expr = ExprParser(
|
field, not stmt.is_doc_based()
|
).parse_table_insert_field()
|
msg["projection"].extend([expr.get_message()])
|
|
for value in stmt.get_values():
|
row = Message("Mysqlx.Crud.Insert.TypedRow")
|
if isinstance(value, list):
|
for val in value:
|
row["field"].extend([build_expr(val).get_message()])
|
else:
|
row["field"].extend([build_expr(value).get_message()])
|
msg["row"].extend([row.get_message()])
|
|
if hasattr(stmt, "is_upsert"):
|
msg["upsert"] = stmt.is_upsert()
|
|
return "Mysqlx.ClientMessages.Type.CRUD_INSERT", msg
|
|
def close_result(self, result: ResultBaseType) -> None:
|
"""Close the result.
|
|
Args:
|
result (Result): A `Result` based type object.
|
|
Raises:
|
:class:`mysqlx.OperationalError`: If message read is None.
|
"""
|
msg = self._read_message(result)
|
if msg is not None:
|
raise OperationalError("Expected to close the result")
|
|
def read_row(self, result: ResultBaseType) -> Optional[MessageType]:
|
"""Read row.
|
|
Args:
|
result (Result): A `Result` based type object.
|
"""
|
msg = self._read_message(result)
|
if msg is None:
|
return None
|
if msg.type == "Mysqlx.Resultset.Row":
|
return msg
|
self._reader.push_message(msg)
|
return None
|
|
def get_column_metadata(self, result: ResultBaseType) -> List[ColumnType]:
|
"""Returns column metadata.
|
|
Args:
|
result (Result): A `Result` based type object.
|
|
Raises:
|
:class:`mysqlx.InterfaceError`: If unexpected message.
|
"""
|
columns = []
|
while True:
|
msg = self._read_message(result)
|
if msg is None:
|
break
|
if msg.type == "Mysqlx.Resultset.Row":
|
self._reader.push_message(msg)
|
break
|
if msg.type != "Mysqlx.Resultset.ColumnMetaData":
|
raise InterfaceError("Unexpected msg type")
|
col = Column(
|
msg["type"],
|
msg["catalog"],
|
msg["schema"],
|
msg["table"],
|
msg["original_table"],
|
msg["name"],
|
msg["original_name"],
|
msg.get("length", 21),
|
msg.get("collation", 0),
|
msg.get("fractional_digits", 0),
|
msg.get("flags", 16),
|
msg.get("content_type"),
|
)
|
columns.append(col)
|
return columns
|
|
def read_ok(self) -> None:
|
"""Read OK.
|
|
Raises:
|
:class:`mysqlx.InterfaceError`: If unexpected message.
|
"""
|
msg = self._reader.read_message()
|
if msg.type == "Mysqlx.Error":
|
raise InterfaceError(f"Mysqlx.Error: {msg['msg']}", errno=msg["code"])
|
if msg.type != "Mysqlx.Ok":
|
raise InterfaceError("Unexpected message encountered")
|
|
def send_connection_close(self) -> None:
|
"""Send connection close."""
|
msg = Message("Mysqlx.Connection.Close")
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.CON_CLOSE"), msg
|
)
|
|
def send_close(self) -> None:
|
"""Send close."""
|
msg = Message("Mysqlx.Session.Close")
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.SESS_CLOSE"), msg
|
)
|
|
def send_expect_open(self) -> None:
|
"""Send expectation."""
|
cond_key = mysqlxpb_enum("Mysqlx.Expect.Open.Condition.Key.EXPECT_FIELD_EXIST")
|
msg_oc = Message("Mysqlx.Expect.Open.Condition")
|
msg_oc["condition_key"] = cond_key
|
msg_oc["condition_value"] = "6.1"
|
|
msg_eo = Message("Mysqlx.Expect.Open")
|
msg_eo["cond"] = [msg_oc.get_message()]
|
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.EXPECT_OPEN"), msg_eo
|
)
|
|
def send_reset(self, keep_open: Optional[bool] = None) -> bool:
|
"""Send reset session message.
|
|
Returns:
|
boolean: ``True`` if the server will keep the session open,
|
otherwise ``False``.
|
"""
|
msg = Message("Mysqlx.Session.Reset")
|
if keep_open is None:
|
try:
|
# Send expectation: keep connection open
|
self.send_expect_open()
|
self.read_ok()
|
keep_open = True
|
except InterfaceError:
|
# Expectation is unkown by this version of the server
|
keep_open = False
|
if keep_open:
|
msg["keep_open"] = True
|
self._writer.write_message(
|
mysqlxpb_enum("Mysqlx.ClientMessages.Type.SESS_RESET"), msg
|
)
|
self.read_ok()
|
if keep_open:
|
return True
|
return False
|
