in mysqlx-connector-python/lib/mysqlx/statement.py [0:0]
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
)