# 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
|
|
# mypy: disable-error-code="return-value"
|
|
"""Implementation of Statements."""
|
|
from __future__ import annotations
|
|
import copy
|
import json
|
import warnings
|
|
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
|
|
from .constants import LockContention
|
from .dbdoc import DbDoc
|
from .errors import NotSupportedError, ProgrammingError
|
from .expr import ExprParser
|
from .helpers import deprecated
|
from .protobuf import mysqlxpb_enum
|
from .result import DocResult, Result, RowResult, SqlResult
|
from .types import (
|
ConnectionType,
|
DatabaseTargetType,
|
MessageType,
|
ProtobufMessageCextType,
|
ProtobufMessageType,
|
SchemaType,
|
)
|
|
ERR_INVALID_INDEX_NAME = 'The given index name "{}" is not valid'
|
|
|
class Expr:
|
"""Expression wrapper."""
|
|
def __init__(self, expr: Any) -> None:
|
self.expr: Any = expr
|
|
|
def flexible_params(*values: Any) -> Union[List, Tuple]:
|
"""Parse flexible parameters."""
|
if len(values) == 1 and isinstance(values[0], (list, tuple)):
|
return values[0]
|
return values
|
|
|
def is_quoted_identifier(identifier: str, sql_mode: str = "") -> bool:
|
"""Check if the given identifier is quoted.
|
|
Args:
|
identifier (string): Identifier to check.
|
sql_mode (Optional[string]): SQL mode.
|
|
Returns:
|
`True` if the identifier has backtick quotes, and False otherwise.
|
"""
|
if "ANSI_QUOTES" in sql_mode:
|
return (identifier[0] == "`" and identifier[-1] == "`") or (
|
identifier[0] == '"' and identifier[-1] == '"'
|
)
|
return identifier[0] == "`" and identifier[-1] == "`"
|
|
|
def quote_identifier(identifier: str, sql_mode: str = "") -> str:
|
"""Quote the given identifier with backticks, converting backticks (`) in
|
the identifier name with the correct escape sequence (``).
|
|
Args:
|
identifier (string): Identifier to quote.
|
sql_mode (Optional[string]): SQL mode.
|
|
Returns:
|
A string with the identifier quoted with backticks.
|
"""
|
if len(identifier) == 0:
|
return "``"
|
if "ANSI_QUOTES" in sql_mode:
|
quoted = identifier.replace('"', '""')
|
return f'"{quoted}"'
|
quoted = identifier.replace("`", "``")
|
return f"`{quoted}`"
|
|
|
def quote_multipart_identifier(identifiers: Iterable[str], sql_mode: str = "") -> str:
|
"""Quote the given multi-part identifier with backticks.
|
|
Args:
|
identifiers (iterable): List of identifiers to quote.
|
sql_mode (Optional[string]): SQL mode.
|
|
Returns:
|
A string with the multi-part identifier quoted with backticks.
|
"""
|
return ".".join(
|
[quote_identifier(identifier, sql_mode) for identifier in identifiers]
|
)
|
|
|
def parse_table_name(
|
default_schema: str, table_name: str, sql_mode: str = ""
|
) -> Tuple[str, str]:
|
"""Parse table name.
|
|
Args:
|
default_schema (str): The default schema.
|
table_name (str): The table name.
|
sql_mode(Optional[str]): The SQL mode.
|
|
Returns:
|
str: The parsed table name.
|
"""
|
quote = '"' if "ANSI_QUOTES" in sql_mode else "`"
|
delimiter = f".{quote}" if quote in table_name else "."
|
temp = table_name.split(delimiter, 1)
|
return (
|
default_schema if len(temp) == 1 else temp[0].strip(quote),
|
temp[-1].strip(quote),
|
)
|
|
|
class Statement:
|
"""Provides base functionality for statement objects.
|
|
Args:
|
target (object): The target database object, it can be
|
:class:`mysqlx.Collection` or :class:`mysqlx.Table`.
|
doc_based (bool): `True` if it is document based.
|
"""
|
|
def __init__(self, target: DatabaseTargetType, doc_based: bool = True) -> None:
|
self._target: DatabaseTargetType = target
|
self._doc_based: bool = doc_based
|
self._connection: Optional[ConnectionType] = (
|
target.get_connection() if target else None
|
)
|
self._stmt_id: Optional[int] = None
|
self._exec_counter: int = 0
|
self._changed: bool = True
|
self._prepared: bool = False
|
self._deallocate_prepare_execute: bool = False
|
|
@property
|
def target(self) -> DatabaseTargetType:
|
"""object: The database object target."""
|
return self._target
|
|
@property
|
def schema(self) -> SchemaType:
|
""":class:`mysqlx.Schema`: The Schema object."""
|
return self._target.schema
|
|
@property
|
def stmt_id(self) -> int:
|
"""Returns this statement ID.
|
|
Returns:
|
int: The statement ID.
|
"""
|
return self._stmt_id
|
|
@stmt_id.setter
|
def stmt_id(self, value: int) -> None:
|
self._stmt_id = value
|
|
@property
|
def exec_counter(self) -> int:
|
"""int: The number of times this statement was executed."""
|
return self._exec_counter
|
|
@property
|
def changed(self) -> bool:
|
"""bool: `True` if this statement has changes."""
|
return self._changed
|
|
@changed.setter
|
def changed(self, value: bool) -> None:
|
self._changed = value
|
|
@property
|
def prepared(self) -> bool:
|
"""bool: `True` if this statement has been prepared."""
|
return self._prepared
|
|
@prepared.setter
|
def prepared(self, value: bool) -> None:
|
self._prepared = value
|
|
@property
|
def repeated(self) -> bool:
|
"""bool: `True` if this statement was executed more than once."""
|
return self._exec_counter > 1
|
|
@property
|
def deallocate_prepare_execute(self) -> bool:
|
"""bool: `True` to deallocate + prepare + execute statement."""
|
return self._deallocate_prepare_execute
|
|
@deallocate_prepare_execute.setter
|
def deallocate_prepare_execute(self, value: bool) -> None:
|
self._deallocate_prepare_execute = value
|
|
def is_doc_based(self) -> bool:
|
"""Check if it is document based.
|
|
Returns:
|
bool: `True` if it is document based.
|
"""
|
return self._doc_based
|
|
def increment_exec_counter(self) -> None:
|
"""Increments the number of times this statement has been executed."""
|
self._exec_counter += 1
|
|
def reset_exec_counter(self) -> None:
|
"""Resets the number of times this statement has been executed."""
|
self._exec_counter = 0
|
|
def execute(self) -> Any:
|
"""Execute the statement.
|
|
Raises:
|
NotImplementedError: This method must be implemented.
|
"""
|
raise NotImplementedError
|
|
|
class FilterableStatement(Statement):
|
"""A statement to be used with filterable statements.
|
|
Args:
|
target (object): The target database object, it can be
|
:class:`mysqlx.Collection` or :class:`mysqlx.Table`.
|
doc_based (Optional[bool]): `True` if it is document based
|
(default: `True`).
|
condition (Optional[str]): Sets the search condition to filter
|
documents or records.
|
"""
|
|
def __init__(
|
self,
|
target: DatabaseTargetType,
|
doc_based: bool = True,
|
condition: Optional[str] = None,
|
) -> None:
|
super().__init__(target=target, doc_based=doc_based)
|
self._binding_map: Dict[str, Any] = {}
|
self._bindings: Union[Dict[str, Any], List] = {}
|
self._having: Optional[MessageType] = None
|
self._grouping_str: str = ""
|
self._grouping: Optional[
|
List[Union[ProtobufMessageType, ProtobufMessageCextType]]
|
] = None
|
self._limit_offset: int = 0
|
self._limit_row_count: int = None
|
self._projection_str: str = ""
|
self._projection_expr: Optional[
|
List[Union[ProtobufMessageType, ProtobufMessageCextType]]
|
] = None
|
self._sort_str: str = ""
|
self._sort_expr: Optional[
|
List[Union[ProtobufMessageType, ProtobufMessageCextType]]
|
] = None
|
self._where_str: str = ""
|
self._where_expr: MessageType = None
|
self.has_bindings: bool = False
|
self.has_limit: bool = False
|
self.has_group_by: bool = False
|
self.has_having: bool = False
|
self.has_projection: bool = False
|
self.has_sort: bool = False
|
self.has_where: bool = False
|
if condition:
|
self._set_where(condition)
|
|
def _bind_single(self, obj: Union[DbDoc, Dict[str, Any], str]) -> None:
|
"""Bind single object.
|
|
Args:
|
obj (:class:`mysqlx.DbDoc` or str): DbDoc or JSON string object.
|
|
Raises:
|
:class:`mysqlx.ProgrammingError`: If invalid JSON string to bind.
|
ValueError: If JSON loaded is not a dictionary.
|
"""
|
if isinstance(obj, dict):
|
self.bind(DbDoc(obj).as_str())
|
elif isinstance(obj, DbDoc):
|
self.bind(obj.as_str())
|
elif isinstance(obj, str):
|
try:
|
res = json.loads(obj)
|
if not isinstance(res, dict):
|
raise ValueError
|
except ValueError as err:
|
raise ProgrammingError("Invalid JSON string to bind") from err
|
for key in res.keys():
|
self.bind(key, res[key])
|
else:
|
raise ProgrammingError("Invalid JSON string or object to bind")
|
|
def _sort(self, *clauses: str) -> FilterableStatement:
|
"""Sets the sorting criteria.
|
|
Args:
|
*clauses: The expression strings defining the sort criteria.
|
|
Returns:
|
mysqlx.FilterableStatement: FilterableStatement object.
|
"""
|
self.has_sort = True
|
self._sort_str = ",".join(flexible_params(*clauses))
|
self._sort_expr = ExprParser(
|
self._sort_str, not self._doc_based
|
).parse_order_spec()
|
self._changed = True
|
return self
|
|
def _set_where(self, condition: str) -> FilterableStatement:
|
"""Sets the search condition to filter.
|
|
Args:
|
condition (str): Sets the search condition to filter documents or
|
records.
|
|
Returns:
|
mysqlx.FilterableStatement: FilterableStatement object.
|
"""
|
self.has_where = True
|
self._where_str = condition
|
try:
|
expr = ExprParser(condition, not self._doc_based)
|
self._where_expr = expr.expr()
|
except ValueError as err:
|
raise ProgrammingError("Invalid condition") from err
|
self._binding_map = expr.placeholder_name_to_position
|
self._changed = True
|
return self
|
|
def _set_group_by(self, *fields: str) -> None:
|
"""Set group by.
|
|
Args:
|
*fields: List of fields.
|
"""
|
fields = flexible_params(*fields)
|
self.has_group_by = True
|
self._grouping_str = ",".join(fields)
|
self._grouping = ExprParser(
|
self._grouping_str, not self._doc_based
|
).parse_expr_list()
|
self._changed = True
|
|
def _set_having(self, condition: str) -> None:
|
"""Set having.
|
|
Args:
|
condition (str): The condition.
|
"""
|
self.has_having = True
|
self._having = ExprParser(condition, not self._doc_based).expr()
|
self._changed = True
|
|
def _set_projection(self, *fields: str) -> FilterableStatement:
|
"""Set the projection.
|
|
Args:
|
*fields: List of fields.
|
|
Returns:
|
:class:`mysqlx.FilterableStatement`: Returns self.
|
"""
|
fields = flexible_params(*fields)
|
self.has_projection = True
|
self._projection_str = ",".join(fields)
|
self._projection_expr = ExprParser(
|
self._projection_str, not self._doc_based
|
).parse_table_select_projection()
|
self._changed = True
|
return self
|
|
def get_binding_map(self) -> Dict[str, Any]:
|
"""Returns the binding map dictionary.
|
|
Returns:
|
dict: The binding map dictionary.
|
"""
|
return self._binding_map
|
|
def get_bindings(self) -> Union[Dict[str, Any], List]:
|
"""Returns the bindings list.
|
|
Returns:
|
`list`: The bindings list.
|
"""
|
return self._bindings
|
|
def get_grouping(self) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
|
"""Returns the grouping expression list.
|
|
Returns:
|
`list`: The grouping expression list.
|
"""
|
return self._grouping
|
|
def get_having(self) -> MessageType:
|
"""Returns the having expression.
|
|
Returns:
|
object: The having expression.
|
"""
|
return self._having
|
|
def get_limit_row_count(self) -> int:
|
"""Returns the limit row count.
|
|
Returns:
|
int: The limit row count.
|
"""
|
return self._limit_row_count
|
|
def get_limit_offset(self) -> int:
|
"""Returns the limit offset.
|
|
Returns:
|
int: The limit offset.
|
"""
|
return self._limit_offset
|
|
def get_where_expr(self) -> MessageType:
|
"""Returns the where expression.
|
|
Returns:
|
object: The where expression.
|
"""
|
return self._where_expr
|
|
def get_projection_expr(
|
self,
|
) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
|
"""Returns the projection expression.
|
|
Returns:
|
object: The projection expression.
|
"""
|
return self._projection_expr
|
|
def get_sort_expr(
|
self,
|
) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
|
"""Returns the sort expression.
|
|
Returns:
|
object: The sort expression.
|
"""
|
return self._sort_expr
|
|
@deprecated("8.0.12")
|
def where(self, condition: str) -> FilterableStatement:
|
"""Sets the search condition to filter.
|
|
Args:
|
condition (str): Sets the search condition to filter documents or
|
records.
|
|
Returns:
|
mysqlx.FilterableStatement: FilterableStatement object.
|
|
.. deprecated:: 8.0.12
|
"""
|
return self._set_where(condition)
|
|
@deprecated("8.0.12")
|
def sort(self, *clauses: str) -> FilterableStatement:
|
"""Sets the sorting criteria.
|
|
Args:
|
*clauses: The expression strings defining the sort criteria.
|
|
Returns:
|
mysqlx.FilterableStatement: FilterableStatement object.
|
|
.. deprecated:: 8.0.12
|
"""
|
return self._sort(*clauses)
|
|
def limit(
|
self, row_count: int, offset: Optional[int] = None
|
) -> FilterableStatement:
|
"""Sets the maximum number of items to be returned.
|
|
Args:
|
row_count (int): The maximum number of items.
|
|
Returns:
|
mysqlx.FilterableStatement: FilterableStatement object.
|
|
Raises:
|
ValueError: If ``row_count`` is not a positive integer.
|
|
.. versionchanged:: 8.0.12
|
The usage of ``offset`` was deprecated.
|
"""
|
if not isinstance(row_count, int) or row_count < 0:
|
raise ValueError("The 'row_count' value must be a positive integer")
|
if not self.has_limit:
|
self._changed = bool(self._exec_counter == 0)
|
self._deallocate_prepare_execute = bool(not self._exec_counter == 0)
|
|
self._limit_row_count = row_count
|
self.has_limit = True
|
if offset:
|
self.offset(offset)
|
warnings.warn(
|
"'limit(row_count, offset)' is deprecated, please "
|
"use 'offset(offset)' to set the number of items to "
|
"skip",
|
category=DeprecationWarning,
|
)
|
return self
|
|
def offset(self, offset: int) -> FilterableStatement:
|
"""Sets the number of items to skip.
|
|
Args:
|
offset (int): The number of items to skip.
|
|
Returns:
|
mysqlx.FilterableStatement: FilterableStatement object.
|
|
Raises:
|
ValueError: If ``offset`` is not a positive integer.
|
|
.. versionadded:: 8.0.12
|
"""
|
if not isinstance(offset, int) or offset < 0:
|
raise ValueError("The 'offset' value must be a positive integer")
|
self._limit_offset = offset
|
return self
|
|
def bind(self, *args: Any) -> FilterableStatement:
|
"""Binds value(s) to a specific placeholder(s).
|
|
Args:
|
*args: The name of the placeholder and the value to bind.
|
A :class:`mysqlx.DbDoc` object or a JSON string
|
representation can be used.
|
|
Returns:
|
mysqlx.FilterableStatement: FilterableStatement object.
|
|
Raises:
|
ProgrammingError: If the number of arguments is invalid.
|
"""
|
self.has_bindings = True
|
count = len(args)
|
if count == 1:
|
self._bind_single(args[0])
|
elif count == 2:
|
self._bindings[args[0]] = args[1]
|
else:
|
raise ProgrammingError("Invalid number of arguments to bind")
|
return self
|
|
def execute(self) -> Any:
|
"""Execute the statement.
|
|
Raises:
|
NotImplementedError: This method must be implemented.
|
"""
|
raise NotImplementedError
|
|
|
class SqlStatement(Statement):
|
"""A statement for SQL execution.
|
|
Args:
|
connection (mysqlx.connection.Connection): Connection object.
|
sql (string): The sql statement to be executed.
|
"""
|
|
def __init__(self, connection: ConnectionType, sql: str) -> None:
|
super().__init__(target=None, doc_based=False)
|
self._connection: ConnectionType = connection
|
self._sql: str = sql
|
self._binding_map: Optional[Dict[str, Any]] = None
|
self._bindings: Union[List, Tuple] = []
|
self.has_bindings: bool = False
|
self.has_limit: bool = False
|
|
@property
|
def sql(self) -> str:
|
"""string: The SQL text statement."""
|
return self._sql
|
|
def get_binding_map(self) -> Dict[str, Any]:
|
"""Returns the binding map dictionary.
|
|
Returns:
|
dict: The binding map dictionary.
|
"""
|
return self._binding_map
|
|
def get_bindings(self) -> Union[Tuple, List]:
|
"""Returns the bindings list.
|
|
Returns:
|
`list`: The bindings list.
|
"""
|
return self._bindings
|
|
def bind(self, *args: Any) -> SqlStatement:
|
"""Binds value(s) to a specific placeholder(s).
|
|
Args:
|
*args: The value(s) to bind.
|
|
Returns:
|
mysqlx.SqlStatement: SqlStatement object.
|
"""
|
if len(args) == 0:
|
raise ProgrammingError("Invalid number of arguments to bind")
|
self.has_bindings = True
|
bindings = flexible_params(*args)
|
if isinstance(bindings, (list, tuple)):
|
self._bindings = bindings
|
else:
|
self._bindings.append(bindings)
|
return self
|
|
def execute(self) -> SqlResult:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.SqlResult: SqlResult object.
|
"""
|
return self._connection.send_sql(self)
|
|
|
class WriteStatement(Statement):
|
"""Provide common write operation attributes."""
|
|
def __init__(self, target: DatabaseTargetType, doc_based: bool) -> None:
|
super().__init__(target, doc_based)
|
self._values: List[
|
Union[
|
int,
|
str,
|
DbDoc,
|
Dict[str, Any],
|
List[Optional[Union[str, int, float, ExprParser, Dict[str, Any]]]],
|
]
|
] = []
|
|
def get_values(
|
self,
|
) -> List[
|
Union[
|
int,
|
str,
|
DbDoc,
|
Dict[str, Any],
|
List[Optional[Union[str, int, float, ExprParser, Dict[str, Any]]]],
|
]
|
]:
|
"""Returns the list of values.
|
|
Returns:
|
`list`: The list of values.
|
"""
|
return self._values
|
|
def execute(self) -> Any:
|
"""Execute the statement.
|
|
Raises:
|
NotImplementedError: This method must be implemented.
|
"""
|
raise NotImplementedError
|
|
|
class AddStatement(WriteStatement):
|
"""A statement for document addition on a collection.
|
|
Args:
|
collection (mysqlx.Collection): The Collection object.
|
"""
|
|
def __init__(self, collection: DatabaseTargetType) -> None:
|
super().__init__(collection, True)
|
self._upsert: bool = False
|
self.ids: List = []
|
|
def is_upsert(self) -> bool:
|
"""Returns `True` if it's an upsert.
|
|
Returns:
|
bool: `True` if it's an upsert.
|
"""
|
return self._upsert
|
|
def upsert(self, value: bool = True) -> AddStatement:
|
"""Sets the upset flag to the boolean of the value provided.
|
Setting of this flag allows updating of the matched rows/documents
|
with the provided value.
|
|
Args:
|
value (optional[bool]): Set or unset the upsert flag.
|
"""
|
self._upsert = value
|
return self
|
|
def add(self, *values: DbDoc) -> AddStatement:
|
"""Adds a list of documents into a collection.
|
|
Args:
|
*values: The documents to be added into the collection.
|
|
Returns:
|
mysqlx.AddStatement: AddStatement object.
|
"""
|
for val in flexible_params(*values):
|
if isinstance(val, DbDoc):
|
self._values.append(val)
|
else:
|
self._values.append(DbDoc(val))
|
return self
|
|
def execute(self) -> Result:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object.
|
"""
|
if len(self._values) == 0:
|
return Result()
|
|
return self._connection.send_insert(self)
|
|
|
class UpdateSpec:
|
"""Update specification class implementation.
|
|
Args:
|
update_type (int): The update type.
|
source (str): The source.
|
value (Optional[str]): The value.
|
|
Raises:
|
ProgrammingError: If `source` is invalid.
|
"""
|
|
def __init__(self, update_type: int, source: str, value: Any = None) -> None:
|
if update_type == mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.SET"):
|
self._table_set(source, value)
|
else:
|
self.update_type: int = update_type
|
try:
|
self.source: Any = ExprParser(source, False).document_field().identifier
|
except ValueError as err:
|
raise ProgrammingError(f"{err}") from err
|
self.value: Any = value
|
|
def _table_set(self, source: str, value: Any) -> None:
|
"""Table set.
|
|
Args:
|
source (str): The source.
|
value (str): The value.
|
"""
|
self.update_type = mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.SET")
|
self.source = ExprParser(source, True).parse_table_update_field()
|
self.value = value
|
|
|
class ModifyStatement(FilterableStatement):
|
"""A statement for document update operations on a Collection.
|
|
Args:
|
collection (mysqlx.Collection): The Collection object.
|
condition (str): Sets the search condition to identify the documents
|
to be modified.
|
|
.. versionchanged:: 8.0.12
|
The ``condition`` parameter is now mandatory.
|
"""
|
|
def __init__(self, collection: DatabaseTargetType, condition: str) -> None:
|
super().__init__(target=collection, condition=condition)
|
self._update_ops: Dict[str, Any] = {}
|
|
def sort(self, *clauses: str) -> ModifyStatement:
|
"""Sets the sorting criteria.
|
|
Args:
|
*clauses: The expression strings defining the sort criteria.
|
|
Returns:
|
mysqlx.ModifyStatement: ModifyStatement object.
|
"""
|
return self._sort(*clauses)
|
|
def get_update_ops(self) -> Dict[str, Any]:
|
"""Returns the list of update operations.
|
|
Returns:
|
`list`: The list of update operations.
|
"""
|
return self._update_ops
|
|
def set(self, doc_path: str, value: Any) -> ModifyStatement:
|
"""Sets or updates attributes on documents in a collection.
|
|
Args:
|
doc_path (string): The document path of the item to be set.
|
value (string): The value to be set on the specified attribute.
|
|
Returns:
|
mysqlx.ModifyStatement: ModifyStatement object.
|
"""
|
self._update_ops[doc_path] = UpdateSpec(
|
mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.ITEM_SET"),
|
doc_path,
|
value,
|
)
|
self._changed = True
|
return self
|
|
@deprecated("8.0.12")
|
def change(self, doc_path: str, value: Any) -> ModifyStatement:
|
"""Add an update to the statement setting the field, if it exists at
|
the document path, to the given value.
|
|
Args:
|
doc_path (string): The document path of the item to be set.
|
value (object): The value to be set on the specified attribute.
|
|
Returns:
|
mysqlx.ModifyStatement: ModifyStatement object.
|
|
.. deprecated:: 8.0.12
|
"""
|
self._update_ops[doc_path] = UpdateSpec(
|
mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.ITEM_REPLACE"),
|
doc_path,
|
value,
|
)
|
self._changed = True
|
return self
|
|
def unset(self, *doc_paths: str) -> ModifyStatement:
|
"""Removes attributes from documents in a collection.
|
|
Args:
|
doc_paths (list): The list of document paths of the attributes to be
|
removed.
|
|
Returns:
|
mysqlx.ModifyStatement: ModifyStatement object.
|
"""
|
for item in flexible_params(*doc_paths):
|
self._update_ops[item] = UpdateSpec(
|
mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.ITEM_REMOVE"),
|
item,
|
)
|
self._changed = True
|
return self
|
|
def array_insert(self, field: str, value: Any) -> ModifyStatement:
|
"""Insert a value into the specified array in documents of a
|
collection.
|
|
Args:
|
field (string): A document path that identifies the array attribute
|
and position where the value will be inserted.
|
value (object): The value to be inserted.
|
|
Returns:
|
mysqlx.ModifyStatement: ModifyStatement object.
|
"""
|
self._update_ops[field] = UpdateSpec(
|
mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.ARRAY_INSERT"),
|
field,
|
value,
|
)
|
self._changed = True
|
return self
|
|
def array_append(self, doc_path: str, value: Any) -> ModifyStatement:
|
"""Inserts a value into a specific position in an array attribute in
|
documents of a collection.
|
|
Args:
|
doc_path (string): A document path that identifies the array
|
attribute and position where the value will be
|
inserted.
|
value (object): The value to be inserted.
|
|
Returns:
|
mysqlx.ModifyStatement: ModifyStatement object.
|
"""
|
self._update_ops[doc_path] = UpdateSpec(
|
mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.ARRAY_APPEND"),
|
doc_path,
|
value,
|
)
|
self._changed = True
|
return self
|
|
def patch(self, doc: Union[Dict, DbDoc, ExprParser, str]) -> ModifyStatement:
|
"""Takes a :class:`mysqlx.DbDoc`, string JSON format or a dict with the
|
changes and applies it on all matching documents.
|
|
Args:
|
doc (object): A generic document (DbDoc), string in JSON format or
|
dict, with the changes to apply to the matching
|
documents.
|
|
Returns:
|
mysqlx.ModifyStatement: ModifyStatement object.
|
"""
|
if doc is None:
|
doc = ""
|
if not isinstance(doc, (ExprParser, dict, DbDoc, str)):
|
raise ProgrammingError(
|
"Invalid data for update operation on document collection table"
|
)
|
self._update_ops["patch"] = UpdateSpec(
|
mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.MERGE_PATCH"),
|
"$",
|
doc.expr() if isinstance(doc, ExprParser) else doc,
|
)
|
self._changed = True
|
return self
|
|
def execute(self) -> Result:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object.
|
|
Raises:
|
ProgrammingError: If condition was not set.
|
"""
|
if not self.has_where:
|
raise ProgrammingError("No condition was found for modify")
|
return self._connection.send_update(self)
|
|
|
class ReadStatement(FilterableStatement):
|
"""Provide base functionality for Read operations
|
|
Args:
|
target (object): The target database object, it can be
|
:class:`mysqlx.Collection` or :class:`mysqlx.Table`.
|
doc_based (Optional[bool]): `True` if it is document based
|
(default: `True`).
|
condition (Optional[str]): Sets the search condition to filter
|
documents or records.
|
"""
|
|
def __init__(
|
self,
|
target: DatabaseTargetType,
|
doc_based: bool = True,
|
condition: Optional[str] = None,
|
) -> None:
|
super().__init__(target, doc_based, condition)
|
self._lock_exclusive: bool = False
|
self._lock_shared: bool = False
|
self._lock_contention: LockContention = LockContention.DEFAULT
|
|
@property
|
def lock_contention(self) -> LockContention:
|
""":class:`mysqlx.LockContention`: The lock contention value."""
|
return self._lock_contention
|
|
def _set_lock_contention(self, lock_contention: LockContention) -> None:
|
"""Set the lock contention.
|
|
Args:
|
lock_contention (:class:`mysqlx.LockContention`): Lock contention.
|
|
Raises:
|
ProgrammingError: If is an invalid lock contention value.
|
"""
|
try:
|
# Check if is a valid lock contention value
|
_ = LockContention(lock_contention.value)
|
except ValueError as err:
|
raise ProgrammingError(
|
"Invalid lock contention mode. Use 'NOWAIT' or 'SKIP_LOCKED'"
|
) from err
|
self._lock_contention = lock_contention
|
|
def is_lock_exclusive(self) -> bool:
|
"""Returns `True` if is `EXCLUSIVE LOCK`.
|
|
Returns:
|
bool: `True` if is `EXCLUSIVE LOCK`.
|
"""
|
return self._lock_exclusive
|
|
def is_lock_shared(self) -> bool:
|
"""Returns `True` if is `SHARED LOCK`.
|
|
Returns:
|
bool: `True` if is `SHARED LOCK`.
|
"""
|
return self._lock_shared
|
|
def lock_shared(
|
self, lock_contention: LockContention = LockContention.DEFAULT
|
) -> ReadStatement:
|
"""Execute a read operation with `SHARED LOCK`. Only one lock can be
|
active at a time.
|
|
Args:
|
lock_contention (:class:`mysqlx.LockContention`): Lock contention.
|
"""
|
self._lock_exclusive = False
|
self._lock_shared = True
|
self._set_lock_contention(lock_contention)
|
return self
|
|
def lock_exclusive(
|
self, lock_contention: LockContention = LockContention.DEFAULT
|
) -> ReadStatement:
|
"""Execute a read operation with `EXCLUSIVE LOCK`. Only one lock can be
|
active at a time.
|
|
Args:
|
lock_contention (:class:`mysqlx.LockContention`): Lock contention.
|
"""
|
self._lock_exclusive = True
|
self._lock_shared = False
|
self._set_lock_contention(lock_contention)
|
return self
|
|
def group_by(self, *fields: str) -> ReadStatement:
|
"""Sets a grouping criteria for the resultset.
|
|
Args:
|
*fields: The string expressions identifying the grouping criteria.
|
|
Returns:
|
mysqlx.ReadStatement: ReadStatement object.
|
"""
|
self._set_group_by(*fields)
|
return self
|
|
def having(self, condition: str) -> ReadStatement:
|
"""Sets a condition for records to be considered in agregate function
|
operations.
|
|
Args:
|
condition (string): A condition on the agregate functions used on
|
the grouping criteria.
|
|
Returns:
|
mysqlx.ReadStatement: ReadStatement object.
|
"""
|
self._set_having(condition)
|
return self
|
|
def execute(self) -> Union[DocResult, RowResult]:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object.
|
"""
|
return self._connection.send_find(self)
|
|
|
class FindStatement(ReadStatement):
|
"""A statement document selection on a Collection.
|
|
Args:
|
collection (mysqlx.Collection): The Collection object.
|
condition (Optional[str]): An optional expression to identify the
|
documents to be retrieved. If not specified
|
all the documents will be included on the
|
result unless a limit is set.
|
"""
|
|
def __init__(
|
self, collection: DatabaseTargetType, condition: Optional[str] = None
|
) -> None:
|
super().__init__(collection, True, condition)
|
|
def fields(self, *fields: str) -> FindStatement:
|
"""Sets a document field filter.
|
|
Args:
|
*fields: The string expressions identifying the fields to be
|
extracted.
|
|
Returns:
|
mysqlx.FindStatement: FindStatement object.
|
"""
|
return self._set_projection(*fields)
|
|
def sort(self, *clauses: str) -> FindStatement:
|
"""Sets the sorting criteria.
|
|
Args:
|
*clauses: The expression strings defining the sort criteria.
|
|
Returns:
|
mysqlx.FindStatement: FindStatement object.
|
"""
|
return self._sort(*clauses)
|
|
|
class SelectStatement(ReadStatement):
|
"""A statement for record retrieval operations on a Table.
|
|
Args:
|
table (mysqlx.Table): The Table object.
|
*fields: The fields to be retrieved.
|
"""
|
|
def __init__(self, table: DatabaseTargetType, *fields: str) -> None:
|
super().__init__(table, False)
|
self._set_projection(*fields)
|
|
def where(self, condition: str) -> SelectStatement:
|
"""Sets the search condition to filter.
|
|
Args:
|
condition (str): Sets the search condition to filter records.
|
|
Returns:
|
mysqlx.SelectStatement: SelectStatement object.
|
"""
|
return self._set_where(condition)
|
|
def order_by(self, *clauses: str) -> SelectStatement:
|
"""Sets the order by criteria.
|
|
Args:
|
*clauses: The expression strings defining the order by criteria.
|
|
Returns:
|
mysqlx.SelectStatement: SelectStatement object.
|
"""
|
return self._sort(*clauses)
|
|
def get_sql(self) -> str:
|
"""Returns the generated SQL.
|
|
Returns:
|
str: The generated SQL.
|
"""
|
where = f" WHERE {self._where_str}" if self.has_where else ""
|
group_by = f" GROUP BY {self._grouping_str}" if self.has_group_by else ""
|
having = f" HAVING {self._having}" if self.has_having else ""
|
order_by = f" ORDER BY {self._sort_str}" if self.has_sort else ""
|
limit = (
|
f" LIMIT {self._limit_row_count} OFFSET {self._limit_offset}"
|
if self.has_limit
|
else ""
|
)
|
stmt = (
|
f"SELECT {self._projection_str or '*'} "
|
f"FROM {self.schema.name}.{self.target.name}"
|
f"{where}{group_by}{having}{order_by}{limit}"
|
)
|
return stmt
|
|
|
class InsertStatement(WriteStatement):
|
"""A statement for insert operations on Table.
|
|
Args:
|
table (mysqlx.Table): The Table object.
|
*fields: The fields to be inserted.
|
"""
|
|
def __init__(self, table: DatabaseTargetType, *fields: Any) -> None:
|
super().__init__(table, False)
|
self._fields: Union[List, Tuple] = flexible_params(*fields)
|
|
def values(self, *values: Any) -> InsertStatement:
|
"""Set the values to be inserted.
|
|
Args:
|
*values: The values of the columns to be inserted.
|
|
Returns:
|
mysqlx.InsertStatement: InsertStatement object.
|
"""
|
self._values.append(list(flexible_params(*values)))
|
return self
|
|
def execute(self) -> Result:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object.
|
"""
|
return self._connection.send_insert(self)
|
|
|
class UpdateStatement(FilterableStatement):
|
"""A statement for record update operations on a Table.
|
|
Args:
|
table (mysqlx.Table): The Table object.
|
|
.. versionchanged:: 8.0.12
|
The ``fields`` parameters were removed.
|
"""
|
|
def __init__(self, table: DatabaseTargetType) -> None:
|
super().__init__(target=table, doc_based=False)
|
self._update_ops: Dict[str, Any] = {}
|
|
def where(self, condition: str) -> UpdateStatement:
|
"""Sets the search condition to filter.
|
|
Args:
|
condition (str): Sets the search condition to filter records.
|
|
Returns:
|
mysqlx.UpdateStatement: UpdateStatement object.
|
"""
|
return self._set_where(condition)
|
|
def order_by(self, *clauses: str) -> UpdateStatement:
|
"""Sets the order by criteria.
|
|
Args:
|
*clauses: The expression strings defining the order by criteria.
|
|
Returns:
|
mysqlx.UpdateStatement: UpdateStatement object.
|
"""
|
return self._sort(*clauses)
|
|
def get_update_ops(self) -> Dict[str, Any]:
|
"""Returns the list of update operations.
|
|
Returns:
|
`list`: The list of update operations.
|
"""
|
return self._update_ops
|
|
def set(self, field: str, value: Any) -> UpdateStatement:
|
"""Updates the column value on records in a table.
|
|
Args:
|
field (string): The column name to be updated.
|
value (object): The value to be set on the specified column.
|
|
Returns:
|
mysqlx.UpdateStatement: UpdateStatement object.
|
"""
|
self._update_ops[field] = UpdateSpec(
|
mysqlxpb_enum("Mysqlx.Crud.UpdateOperation.UpdateType.SET"),
|
field,
|
value,
|
)
|
self._changed = True
|
return self
|
|
def execute(self) -> Result:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object
|
|
Raises:
|
ProgrammingError: If condition was not set.
|
"""
|
if not self.has_where:
|
raise ProgrammingError("No condition was found for update")
|
return self._connection.send_update(self)
|
|
|
class RemoveStatement(FilterableStatement):
|
"""A statement for document removal from a collection.
|
|
Args:
|
collection (mysqlx.Collection): The Collection object.
|
condition (str): Sets the search condition to identify the documents
|
to be removed.
|
|
.. versionchanged:: 8.0.12
|
The ``condition`` parameter was added.
|
"""
|
|
def __init__(self, collection: DatabaseTargetType, condition: str) -> None:
|
super().__init__(target=collection, condition=condition)
|
|
def sort(self, *clauses: str) -> RemoveStatement:
|
"""Sets the sorting criteria.
|
|
Args:
|
*clauses: The expression strings defining the sort criteria.
|
|
Returns:
|
mysqlx.FindStatement: FindStatement object.
|
"""
|
return self._sort(*clauses)
|
|
def execute(self) -> Result:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object.
|
|
Raises:
|
ProgrammingError: If condition was not set.
|
"""
|
if not self.has_where:
|
raise ProgrammingError("No condition was found for remove")
|
return self._connection.send_delete(self)
|
|
|
class DeleteStatement(FilterableStatement):
|
"""A statement that drops a table.
|
|
Args:
|
table (mysqlx.Table): The Table object.
|
|
.. versionchanged:: 8.0.12
|
The ``condition`` parameter was removed.
|
"""
|
|
def __init__(self, table: DatabaseTargetType) -> None:
|
super().__init__(target=table, doc_based=False)
|
|
def where(self, condition: str) -> DeleteStatement:
|
"""Sets the search condition to filter.
|
|
Args:
|
condition (str): Sets the search condition to filter records.
|
|
Returns:
|
mysqlx.DeleteStatement: DeleteStatement object.
|
"""
|
return self._set_where(condition)
|
|
def order_by(self, *clauses: str) -> DeleteStatement:
|
"""Sets the order by criteria.
|
|
Args:
|
*clauses: The expression strings defining the order by criteria.
|
|
Returns:
|
mysqlx.DeleteStatement: DeleteStatement object.
|
"""
|
return self._sort(*clauses)
|
|
def execute(self) -> Result:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object.
|
|
Raises:
|
ProgrammingError: If condition was not set.
|
"""
|
if not self.has_where:
|
raise ProgrammingError("No condition was found for delete")
|
return self._connection.send_delete(self)
|
|
|
class CreateCollectionIndexStatement(Statement):
|
"""A statement that creates an index on a collection.
|
|
Args:
|
collection (mysqlx.Collection): Collection.
|
index_name (string): Index name.
|
index_desc (dict): A dictionary containing the fields members that
|
constraints the index to be created. It must have
|
the form as shown in the following::
|
|
{"fields": [{"field": member_path,
|
"type": member_type,
|
"required": member_required,
|
"collation": collation,
|
"options": options,
|
"srid": srid},
|
# {... more members,
|
# repeated as many times
|
# as needed}
|
],
|
"type": type}
|
"""
|
|
def __init__(
|
self,
|
collection: DatabaseTargetType,
|
index_name: str,
|
index_desc: Dict[str, Any],
|
) -> None:
|
super().__init__(target=collection)
|
self._index_desc: Dict[str, Any] = copy.deepcopy(index_desc)
|
self._index_name: str = index_name
|
self._fields_desc: List[Dict[str, Any]] = self._index_desc.pop("fields", [])
|
|
def execute(self) -> Result:
|
"""Execute the statement.
|
|
Returns:
|
mysqlx.Result: Result object.
|
"""
|
# Validate index name is a valid identifier
|
if self._index_name is None:
|
raise ProgrammingError(ERR_INVALID_INDEX_NAME.format(self._index_name))
|
try:
|
parsed_ident = ExprParser(self._index_name).expr().get_message()
|
|
# The message is type dict when the Protobuf cext is used
|
if isinstance(parsed_ident, dict):
|
if parsed_ident["type"] != mysqlxpb_enum("Mysqlx.Expr.Expr.Type.IDENT"):
|
raise ProgrammingError(
|
ERR_INVALID_INDEX_NAME.format(self._index_name)
|
)
|
else:
|
if parsed_ident.type != mysqlxpb_enum("Mysqlx.Expr.Expr.Type.IDENT"):
|
raise ProgrammingError(
|
ERR_INVALID_INDEX_NAME.format(self._index_name)
|
)
|
|
except (ValueError, AttributeError) as err:
|
raise ProgrammingError(
|
ERR_INVALID_INDEX_NAME.format(self._index_name)
|
) from err
|
|
# Validate members that constraint the index
|
if not self._fields_desc:
|
raise ProgrammingError(
|
"Required member 'fields' not found in the given index "
|
f"description: {self._index_desc}"
|
)
|
|
if not isinstance(self._fields_desc, list):
|
raise ProgrammingError("Required member 'fields' must contain a list")
|
args: Dict[str, Any] = {}
|
args["name"] = self._index_name
|
args["collection"] = self._target.name
|
args["schema"] = self._target.schema.name
|
if "type" in self._index_desc:
|
args["type"] = self._index_desc.pop("type")
|
else:
|
args["type"] = "INDEX"
|
args["unique"] = self._index_desc.pop("unique", False)
|
# Currently unique indexes are not supported:
|
if args["unique"]:
|
raise NotSupportedError("Unique indexes are not supported.")
|
args["constraint"] = []
|
|
if self._index_desc:
|
raise ProgrammingError(f"Unidentified fields: {self._index_desc}")
|
|
try:
|
for field_desc in self._fields_desc:
|
constraint = {}
|
constraint["member"] = field_desc.pop("field")
|
constraint["type"] = field_desc.pop("type")
|
constraint["required"] = field_desc.pop("required", False)
|
constraint["array"] = field_desc.pop("array", False)
|
if not isinstance(constraint["required"], bool):
|
raise TypeError("Field member 'required' must be Boolean")
|
if not isinstance(constraint["array"], bool):
|
raise TypeError("Field member 'array' must be Boolean")
|
if args["type"].upper() == "SPATIAL" and not constraint["required"]:
|
raise ProgrammingError(
|
"Field member 'required' must be set to 'True' when "
|
"index type is set to 'SPATIAL'"
|
)
|
if args["type"].upper() == "INDEX" and constraint["type"] == "GEOJSON":
|
raise ProgrammingError(
|
"Index 'type' must be set to 'SPATIAL' when field "
|
"type is set to 'GEOJSON'"
|
)
|
if "collation" in field_desc:
|
if not constraint["type"].upper().startswith("TEXT"):
|
raise ProgrammingError(
|
"The 'collation' member can only be used when "
|
"field type is set to "
|
f"'{constraint['type'].upper()}'"
|
)
|
constraint["collation"] = field_desc.pop("collation")
|
# "options" and "srid" fields in IndexField can be
|
# present only if "type" is set to "GEOJSON"
|
if "options" in field_desc:
|
if constraint["type"].upper() != "GEOJSON":
|
raise ProgrammingError(
|
"The 'options' member can only be used when "
|
"index type is set to 'GEOJSON'"
|
)
|
constraint["options"] = field_desc.pop("options")
|
if "srid" in field_desc:
|
if constraint["type"].upper() != "GEOJSON":
|
raise ProgrammingError(
|
"The 'srid' member can only be used when index "
|
"type is set to 'GEOJSON'"
|
)
|
constraint["srid"] = field_desc.pop("srid")
|
args["constraint"].append(constraint)
|
except KeyError as err:
|
raise ProgrammingError(
|
f"Required inner member {err} not found in constraint: {field_desc}"
|
) from err
|
|
for field_desc in self._fields_desc:
|
if field_desc:
|
raise ProgrammingError(f"Unidentified inner fields: {field_desc}")
|
|
return self._connection.execute_nonquery(
|
"mysqlx", "create_collection_index", True, args
|
)
|
