# Copyright (c) 2022, 2025, 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 designed to work 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 either included with # the program or referenced in the documentation. # # 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 from typing import Literal, Optional from mrs_plugin import lib from pathlib import Path import os import re from string import Template import json from base64 import b64decode import mysqlsh SDK_PYTHON_DATACLASS_TEMPLATE = '''@dataclass(init=False, repr=True) class I{name}({mixins}MrsDocument[I{name}Data] ): # For data attributes, `None` means "NULL" and # `UndefinedField` means "not set or undefined" {join_field_block} def __init__( # type: ignore[override] self, schema: MrsBaseSchema, data: I{name}Data ) -> None: super().__init__( schema, data, fields_map={field_profile}, obj_endpoint="{obj_endpoint}", ) @classmethod def get_primary_key_name(cls) -> Optional[str]: return {primary_key_name} ''' SDK_PYTHON_NON_NATIVE_TYPES = ("Date", "DateTime", "Time", "Year", "Vector") class LanguageNotSupportedError(Exception): def __init__(self, sdk_language): self.message = f"The SDK language {sdk_language} is not supported yet." def get_base_classes(sdk_language, prepare_for_runtime=False): if sdk_language == "TypeScript": file_name = "MrsBaseClasses.ts" elif sdk_language == "Python": file_name = "mrs_base_classes.py" else: raise LanguageNotSupportedError(sdk_language) path = os.path.abspath(__file__) code = Path(os.path.dirname(path), "..", "sdk", sdk_language.lower(), file_name).read_text() if prepare_for_runtime is True: # Remove exports as everything will be in a single file code = code.replace("export ", "") # Remove the part that does not belong in the runtime SDK delimiter = language_comment_delimiter(sdk_language) code = re.sub(f"^\\s*?{delimiter} --- MySQL Shell for VS Code Extension Remove --- Begin.*?" + f"^\\s*?{delimiter} --- MySQL Shell for VS Code Extension Remove --- End\n", "", code, flags=re.DOTALL | re.MULTILINE) code = remove_js_whitespace_and_comments(code) return code def generate_service_sdk(service, sdk_language, session, prepare_for_runtime=False, service_url=None): # If no services is given, return only the MRS runtime management TypeScript Code # that allows the user to manage specific MRS runtime settings if service is None: if prepare_for_runtime is True and sdk_language == "TypeScript": return remove_js_whitespace_and_comments(get_mrs_runtime_management_code(session=session)) return "" if sdk_language == "TypeScript": file_name = "MrsServiceTemplate.ts.template" elif sdk_language == "Python": file_name = "mrs_service_template.py.template" else: raise LanguageNotSupportedError(sdk_language) path = os.path.abspath(__file__) template = Path(os.path.dirname(path), "..", "sdk", sdk_language.lower(), file_name).read_text() def binary_formatter(base64str: str): if base64str.startswith("type254:"): return b64decode(base64str.split(":")[-1]) # booleans are encoded as BIT(1) if base64str.startswith("type16:"): return bool.from_bytes(b64decode(base64str.split(":")[-1])) return base64str try: sdk_data = lib.services.get_service_sdk_data( session, service.get("id"), binary_formatter=binary_formatter ) except mysqlsh.DBError as e: if e.code != 1370: raise e # we should make a clear distinction for the case where we cannot retrieve # sdk data because the user is missing "EXECUTE" permissions (BUG#) sdk_data = None # Process Template String code = substitute_service_in_template( service=service, template=template, sdk_language=sdk_language, session=session, service_url=service_url, # service_data should be None if it couldn't be retrieved beforehand (BUG#) # otherwise, if there are no services, it should be an empty dictionary service_data=None if sdk_data is None else sdk_data.get("service_res", {}), ) code = substitute_imports_in_template( template=code.get("template"), enabled_crud_ops=code.get("enabled_crud_ops"), required_datatypes=code.get("required_datatypes"), sdk_language=sdk_language, requires_auth=code.get("requires_auth", False) ) template = code.get("template") delimiter = language_comment_delimiter(sdk_language) if prepare_for_runtime is True: # Remove imports as everything will be in a single file template = re.sub('import.*?;', '', template, flags=re.DOTALL | re.MULTILINE) # Remove exports as everything will be in a single file template = template.replace("export ", "") # Remove the part that does not belong in the runtime SDK template = re.sub(f"^[^\\S\r\n]*?{delimiter} --- MySQL Shell for VS Code Extension Remove --- Begin.*?" + f"^\\s*?{delimiter} --- MySQL Shell for VS Code Extension Remove --- End\n", "", template, flags=re.DOTALL | re.MULTILINE) # Add MRS management code template += get_mrs_runtime_management_code(session=session, service_url=service_url) template = remove_js_whitespace_and_comments(template) else: # Remove the part that does not belong in the generated SDK template = re.sub(f"^[^\\S\r\n]*?{delimiter} --- MySQL Shell for VS Code Extension Only --- Begin.*?" + f"^\\s*?{delimiter} --- MySQL Shell for VS Code Extension Only --- End\n", "", template, flags=re.DOTALL | re.MULTILINE) return template def substitute_imports_in_template( template, sdk_language, enabled_crud_ops: set[str] = set(), required_datatypes: set[str] = set(), requires_auth: bool = False, ): delimiter = language_comment_delimiter(sdk_language) import_loops = re.finditer( f"^[^\\S\r\n]*?{delimiter} --- importLoopStart\n\\s*(^[\\S\\s]*?)^\\s*?{delimiter} --- importLoopEnd\n", template, flags=re.DOTALL | re.MULTILINE, ) ops = [ "Create", "Read", "Update", "Delete", "ProcedureCall", "ReadUnique", "FunctionCall", "Authenticate", "TaskRun", ] enabled_ops = enabled_crud_ops if enabled_crud_ops else set() task_ops = {"FunctionTaskRun", "ProcedureTaskRun"} if len(enabled_ops) - len(enabled_ops.difference(task_ops)) > 0: enabled_ops = enabled_ops.difference(task_ops) enabled_ops.add("TaskRun") if requires_auth: enabled_ops.add("Authenticate") for loop in import_loops: import_template = loop.group(1) # if there are no required datatypes, we should remove the template tags datatypes_block = "" # otherwise we should replace them with the corresponding import block if len(required_datatypes) > 0: # tab size to be converted to spaces indent = " " * 4 # each datatype should be comma separated and belongs in a new line separator = f",\n{indent}" # The first and last datatypes should also follow the same rules and they should be sorted alphabetically, # mostly for testing purposes, but it is always good to be deterministic. datatypes_block = f"{indent}{separator.join(sorted(required_datatypes))},\n" import_template = re.sub( f"^[^\\S\r\n]*?{delimiter} --- importRequiredDatatypesOnlyStart.*?" + f"^\\s*?{delimiter} --- importRequiredDatatypesOnlyEnd\n", datatypes_block, import_template, flags=re.DOTALL | re.MULTILINE, ) for op in ops: # Find all "import{crud_op}OnlyStart / End" blocks op_loops = re.finditer( f"^[^\\S\r\n]*?{delimiter} --- import{op}OnlyStart\n\\s*(^[\\S\\s]*?)^\\s*?{delimiter} --- import{op}OnlyEnd\n", import_template, flags=re.DOTALL | re.MULTILINE, ) for op_loop in op_loops: # If the `CRUD + Auth` operation is enabled for any DB Object, keep the identified code block if op in enabled_ops: import_template = import_template.replace( op_loop.group(), op_loop.group(1) ) else: # Delete the identified code block otherwise # This cannot happen, because the block is gone for the next db object import_template = import_template.replace(op_loop.group(), "") template = template.replace(loop.group(), import_template) return { "template": template, "enabled_crud_ops": enabled_crud_ops, "required_datatypes": required_datatypes, } def substitute_service_in_template(service, template, sdk_language, session, service_url, service_data): # Currently, we only generate the SDK for a single service, but this might change in the future. existing_identifiers = [] code = substitute_schemas_in_template( service=service, template=template, sdk_language=sdk_language, session=session, service_url=service_url, schemas=None if service_data is None else service_data.get("db_schemas", []), ) template = code.get("template") requires_auth = code.get("requires_auth") # if no db object requires auth and there are no authentication apps enabled for the service, # the SDK authentication command should not be generated delimiter = language_comment_delimiter(sdk_language) auth_loops = re.finditer( pattern=( "^[^\\S\r\n]*?" f"{delimiter} --- serviceAuthenticateStart" "\n\\s*(^[\\S\\s]*?)^\\s*?" f"{delimiter} --- serviceAuthenticateEnd\n" ), string=template, flags=re.DOTALL | re.MULTILINE ) for auth_loop in auth_loops: if requires_auth: template = template.replace(auth_loop.group(), auth_loop.group(1)) else: template = template.replace(auth_loop.group(), "") service_id = service.get("id") mapping = { "service_name": generate_identifier( value=service.get("url_context_root"), sdk_language=sdk_language, existing_identifiers=existing_identifiers, ), "service_class_name": generate_identifier( value=service.get("url_context_root"), primitive="class", existing_identifiers=existing_identifiers, ), "service_url": service_url, "service_auth_path": service.get("auth_path"), "service_id": lib.core.convert_id_to_string(service_id), } template = Template(template).substitute(**mapping) return {"template": template, "enabled_crud_ops": code.get("enabled_crud_ops"), "required_datatypes": code.get("required_datatypes"), "requires_auth": requires_auth} def substitute_schemas_in_template( service, template, sdk_language, session, service_url, schemas ): delimiter = language_comment_delimiter(sdk_language) schema_loops = re.finditer( f"^[^\\S\r\n]*?{delimiter} --- schemaLoopStart\n\\s*(^[\\S\\s]*?)^\\s*?{delimiter} --- schemaLoopEnd\n", template, flags=re.DOTALL | re.MULTILINE, ) # BUG#37926204 Get database schemas if they have not been retrieved beforehand db_schemas = lib.schemas.query_schemas(session, service_id=service.get("id")) if schemas is None else schemas enabled_crud_ops = set() required_datatypes = set() requires_auth = False for loop in schema_loops: # for each schema loop, we should restart tracking the identifiers existing_identifiers = [] schema_template = loop.group(1) filled_temp = "" for schema in db_schemas: # TODO: Implement support for MRS Scripts if schema.get("schema_type") == "SCRIPT_MODULE": continue if requires_auth is False: requires_auth |= schema.get("requires_auth") == 1 delimiter = language_comment_delimiter(sdk_language) if f"{delimiter} --- objectLoopStart" in schema_template: # Fill inner Object loops code = substitute_objects_in_template( service=service, schema=schema, template=schema_template, sdk_language=sdk_language, session=session, service_url=service_url, db_objs=None if schemas is None else schema.get("db_objects", []) ) schema_template_with_obj_filled = code.get("template") enabled_crud_ops.update(code.get("enabled_crud_ops")) required_datatypes.update(code.get("required_datatypes")) if requires_auth is False: requires_auth |= code.get("requires_auth") else: schema_template_with_obj_filled = schema_template mapping = { "schema_name": generate_identifier( value=schema.get("request_path"), sdk_language=sdk_language, existing_identifiers=existing_identifiers, ), "schema_class_name": generate_identifier( value=f"{service.get("url_context_root")}{schema.get("request_path")}", primitive="class", existing_identifiers=existing_identifiers, ), "schema_request_path": schema.get("request_path"), "schema_id": lib.core.convert_id_to_string(schema.get("id")), } filled_temp += Template(schema_template_with_obj_filled).substitute( **mapping ) template = template.replace(loop.group(), filled_temp) return { "template": template, "enabled_crud_ops": enabled_crud_ops, "required_datatypes": required_datatypes, "requires_auth": requires_auth, } def get_mrs_object_sdk_language_options(sdk_options, sdk_language): if not sdk_options or not sdk_options.get("language_options"): return {} # Find the matching language_options for opt in sdk_options.get("language_options"): if opt.get("language") == sdk_language: return opt return {} def language_comment_delimiter(sdk_language): if sdk_language == "TypeScript": return "//" if sdk_language == "Python": return "#" def generate_identifier( value: str, primitive: Literal["variable", "class"] = "variable", sdk_language: Literal["TypeScript", "Python"] = "TypeScript", # mutable objects for default values are re-used on subsequent function calls, let's leverage that existing_identifiers: list[str] = [], allowed_special_characters: Optional[set[str]] = None ) -> str: if primitive == "class": identifier = f"{lib.core.convert_path_to_pascal_case(value, allowed_special_characters)}" elif sdk_language == "TypeScript": identifier = f"{lib.core.convert_path_to_camel_case(value, allowed_special_characters)}" elif sdk_language == "Python": identifier = f"{lib.core.convert_to_snake_case(lib.core.convert_path_to_camel_case(value, allowed_special_characters))}" else: identifier = value identifier = f"_{identifier}" if identifier[0].isdigit() else identifier total_duplicates = existing_identifiers.count(identifier) # we want to track the identifier with a potential prefix, but without the suffix existing_identifiers.append(identifier) # the total number of duplicates determines the suffix identifier = ( identifier if total_duplicates == 0 else f"{identifier}{total_duplicates}" ) return identifier def substitute_objects_in_template( service, schema, template, sdk_language, session, service_url, db_objs ): delimiter = language_comment_delimiter(sdk_language) object_loops = re.finditer( f"^[^\\S\r\n]*?{delimiter} --- objectLoopStart\n\\s*(^[\\S\\s]*?)^\\s*?{delimiter} --- objectLoopEnd\n", template, flags=re.DOTALL | re.MULTILINE, ) # BUG#37926204 Get database objects if they have not been retrieved beforehand db_objects = lib.db_objects.query_db_objects(session, schema_id=schema.get("id")) if db_objs is None else db_objs crud_ops = [ "Create", "Read", "Update", "Delete", "DeleteUnique", "ProcedureCall", "ReadUnique", "FunctionCall", "FunctionTaskRun", "ProcedureTaskRun" ] enabled_crud_ops = set() required_datatypes = set() requires_auth = False schema_request_path = f"{service.get("url_context_root")}{schema.get("request_path")}" schema_class_name = generate_identifier( value=schema_request_path, primitive="class", existing_identifiers=[], ) for loop in object_loops: # for each object loop, we need to restart tracking the identifiers existing_identifiers = [] filled_temp = "" for db_obj in db_objects: name = generate_identifier( value=db_obj.get("request_path"), sdk_language=sdk_language, existing_identifiers=existing_identifiers, ) class_name = "" obj_class_name = generate_identifier( value=f"{schema_request_path}{db_obj.get("request_path")}", primitive="class", existing_identifiers=existing_identifiers, ) obj_interfaces = "" obj_meta_interface = "I" + obj_class_name + "ResultSet" obj_param_interface = "" getters_setters = "" obj_pk_list = [] obj_quoted_pk_list = [] obj_string_pk_list = [] obj_string_args_where_pk_list = [] obj_unique_list = [] obj_meta_interfaces = [] db_object_crud_ops = "" # BUG#37926204 Get SDK objects if they have not been retrieved beforehand objects = lib.db_objects.get_objects( session, db_object_id=db_obj.get("id")) if db_obj.get("objects") is None else db_obj.get("objects") # Loop over all objects and build interfaces for obj in objects: if requires_auth is False: requires_auth |= db_obj.get("requires_auth") == 1 # BUG#37926204 Get object fields if they have not been retrieved beforehand fields = lib.db_objects.get_object_fields_with_references( session=session, object_id=obj.get("id")) if obj.get("fields") is None else obj.get("fields") for field in fields: if field.get("lev") == 1: # Build Primary Key lists (only if "UPDATE" is allowed) if field_is_pk(field) and "UPDATE" in db_obj.get("crud_operations", []): obj_pk_list.append(field.get("name")) obj_quoted_pk_list.append(f'"{field.get("name")}"') obj_string_pk_list.append( f'String({name}.{field.get("name")})') obj_string_args_where_pk_list.append( f'String(args.where.{field.get("name")})') # Build Unique list if field_is_unique(field): obj_unique_list.append(field.get("name")) # Build list containing the required types to import from MrsBaseClasses.ts db_column_info = field.get("db_column") if db_column_info: db_datatype = db_column_info.get("datatype") db_not_null = db_column_info.get("not_null") # In TypeScript, there is no native type declaration for SomeType | null, so we add # our own. # In Python, we have Optional[SomeType] that does the trick, so there there is no # need to add one. if db_not_null is False and sdk_language == "TypeScript": required_datatypes.add("MaybeNull") client_datatype = get_enhanced_datatype_mapping(db_datatype, sdk_language) if not datatype_is_primitive(client_datatype, sdk_language): required_datatypes.add(client_datatype) if ( sdk_language == "Python" and client_datatype.startswith( SDK_PYTHON_NON_NATIVE_TYPES ) ): required_datatypes.add( client_datatype.replace("Field", "") ) # Get sdk_language specific options sdk_lang_options = get_mrs_object_sdk_language_options( obj.get("sdk_options"), sdk_language) # Either take the custom interface_name or the default class_name class_name = sdk_lang_options.get( "class_name", obj_class_name) # For database objects other than PROCEDUREs and FUNCTIONS, if there are unique fields, # the corresponding SDK commands should be enabled. if not object_is_routine(db_obj): # READ is always enabled db_object_crud_ops = db_obj.get("crud_operations", "READ") # If db_objects results from calling lib.db_objects.get_objects, the value of the "crud_operations" # key is already a list, if it results from calling lib.services.get_service_sdk_data, the value # is a comma-separated string (BUG#37926204) db_object_crud_ops = db_object_crud_ops if isinstance(db_object_crud_ops, list) else db_object_crud_ops.split(",") # If this DB Object has unique columns (PK or UNIQUE) allow ReadUnique if len(obj_unique_list) > 0 and "READUNIQUE" not in db_object_crud_ops: db_object_crud_ops.append("READUNIQUE") if len(obj_unique_list) > 0 and "DELETE" in db_object_crud_ops and "DELETEUNIQUE" not in db_object_crud_ops: db_object_crud_ops.append("DELETEUNIQUE") # If the database object is a FUNCTION a PROCEDURE or a SCRIPT, CRUD operations should not be enabled elif object_is_routine(db_obj, of_type={"FUNCTION", "SCRIPT"}): options = db_obj.get("options") if options is not None and options.get("mysqlTask") is not None: db_object_crud_ops = ["FUNCTIONTASKRUN"] else: db_object_crud_ops = ["FUNCTIONCALL"] else: required_datatypes.add("IMrsProcedureResult") options = db_obj.get("options") if options is not None and options.get("mysqlTask") is not None: db_object_crud_ops = ["PROCEDURETASKRUN"] else: db_object_crud_ops = ["PROCEDURECALL"] obj_interfaces_def, required_obj_datatypes = generate_interfaces( db_obj, obj, fields, class_name, sdk_language, db_object_crud_ops, obj_endpoint=f"{service_url}{schema.get('request_path')}{db_obj.get("request_path")}", ) required_datatypes.update(required_obj_datatypes) # Do not add obj_interfaces for FUNCTION results if obj.get("kind") == "PARAMETERS" or not object_is_routine(db_obj, of_type={"FUNCTION"}): obj_interfaces += obj_interfaces_def if obj.get("kind") == "PARAMETERS" and object_is_routine(db_obj): obj_param_interface = f"{class_name}Params" if obj.get("kind") != "PARAMETERS" and object_is_routine(db_obj): obj_meta_interfaces.append(class_name) # If the db object is a function, get the return datatype obj_function_result_datatype = None if object_is_routine(db_obj, of_type={"FUNCTION"}): obj_function_result_datatype = "unknown" if len(objects) > 1: # BUG#37926204 Get object fields if they have not been retrieved beforehand fields = lib.db_objects.get_object_fields_with_references( session=session, object_id=objects[1].get("id")) if objects[1].get("fields") is None else objects[1].get("fields") if len(fields) > 0: db_column_info = field.get("db_column") if db_column_info: obj_function_result_datatype = get_datatype_mapping( db_datatype=db_column_info.get("datatype"), sdk_language=sdk_language) # If there are no typed result sets for a Procedure, all the result sets will be generic instances of JsonObject obj_procedure_result_set_datatype = None if object_is_routine(db_obj, of_type={"PROCEDURE"}) and len(objects) == 1: required_datatypes.add("JsonObject") if sdk_language != "Python": obj_interfaces += generate_union( obj_meta_interface, ["JsonObject"], sdk_language ) else: obj_interfaces += generate_union( obj_meta_interface, ["MrsProcedureResultSet[str, JsonObject, JsonObject]"], sdk_language, ) elif object_is_routine(db_obj, of_type={"PROCEDURE"}): if sdk_language != "Python": # TypeScript tagged unions inherit from JsonObject required_datatypes.add("JsonObject") interface_list = [f"ITagged{name}" for name in obj_meta_interfaces] else: interface_list = [ f"MrsProcedureResultSet[I{name}Type, I{name}, ITagged{name}]" for name in obj_meta_interfaces ] obj_procedure_result_set_datatype = ( "{" + ",".join( [f'"{name}": I{name}' for name in obj_meta_interfaces] ) + "}" ) obj_interfaces += generate_union(obj_meta_interface, interface_list, sdk_language) # Define the mappings mapping = { "obj_id": lib.core.convert_id_to_string(db_obj.get("id")), "obj_name": name, "obj_class_name": class_name, "obj_param_interface": obj_param_interface, # empty if not FUNCTION/PROCEDURE "obj_meta_interface": obj_meta_interface, "obj_request_path": db_obj.get("request_path"), "schema_class_name": schema_class_name, "schema_request_path": schema.get("request_path"), "obj_full_request_path": service.get("url_context_root") + schema.get("request_path") + db_obj.get("request_path"), "obj_type": db_obj.get("object_type"), "obj_interfaces": obj_interfaces, "obj_getters_setters": getters_setters, "obj_pk_list": ", ".join(obj_pk_list), "obj_quoted_pk_list": ", ".join(obj_quoted_pk_list), "obj_string_pk_list": ", ".join(obj_string_pk_list), "obj_string_args_where_pk_list": ", ".join( obj_string_args_where_pk_list ), "obj_function_result_datatype": obj_function_result_datatype, "obj_procedure_result_set_datatype": obj_procedure_result_set_datatype, } # Loop over all CRUD operations and filter the sections that are not applicable for the specific object obj_template = loop.group(1) for crud_op in crud_ops: # Find all crud{crud_op}OnlyStart / End control blocks delimiter = language_comment_delimiter(sdk_language) crud_op_loops = re.finditer( f"^[^\\S\r\n]*?{delimiter} --- crud{crud_op}OnlyStart\n\\s*(^[\\S\\s]*?)^\\s*?{delimiter} --- crud{crud_op}OnlyEnd\n", obj_template, flags=re.DOTALL | re.MULTILINE) for crud_loop in crud_op_loops: # If the CRUD operation is enabled for this DB Object, keep the identified code block # if crud_op is "Update" and there is no primary key, update commands should not be available if crud_op.upper() in db_object_crud_ops and (crud_op != "Update" or len(obj_pk_list) > 0): enabled_crud_ops.add(crud_op) obj_template = obj_template.replace( crud_loop.group(), crud_loop.group(1)) else: # Delete the identified code block otherwise obj_template = obj_template.replace( crud_loop.group(), "") # Perform the substitution filled_temp += Template(obj_template).substitute(**mapping) template = template.replace(loop.group(), filled_temp) return {"template": template, "enabled_crud_ops": enabled_crud_ops, "required_datatypes": required_datatypes, "requires_auth": requires_auth} def get_datatype_mapping(db_datatype, sdk_language): if db_datatype is None: db_datatype = "text" db_datatype = db_datatype.lower() if sdk_language == "TypeScript": if db_datatype.startswith(("tinyint(1)", "bit(1)")): return "boolean" if db_datatype.startswith(("tinyint", "smallint", "mediumint", "int", "bigint", "decimal", "numeric", "float", "double")): return "number" if db_datatype.startswith("json"): return "JsonValue" if db_datatype.startswith("geometrycollection"): return "GeometryCollection" if db_datatype.startswith("geometry"): return "Geometry" if db_datatype.startswith("point"): return "Point" if db_datatype.startswith("multipoint"): return "MultiPoint" if db_datatype.startswith("linestring"): return "LineString" if db_datatype.startswith("multilinestring"): return "MultiLineString" if db_datatype.startswith("polygon"): return "Polygon" if db_datatype.startswith("multipolygon"): return "MultiPolygon" return "string" if sdk_language == "Python": if db_datatype.startswith(("tinyint(1)", "bit(1)")): return "bool" if db_datatype.startswith(("tinyint", "smallint", "mediumint", "int", "bigint")): return "int" if db_datatype.startswith(("decimal", "numeric", "float", "double")): return "float" if db_datatype.startswith("json"): return "JsonValue" if db_datatype.startswith("geometry"): return "Geometry" if db_datatype.startswith("point"): return "Point" if db_datatype.startswith("linestring"): return "LineString" if db_datatype.startswith("polygon"): return "Polygon" if db_datatype.startswith("multipoint"): return "MultiPoint" if db_datatype.startswith("multilinestring"): return "MultiLineString" if db_datatype.startswith("multipolygon"): return "MultiPolygon" if db_datatype.startswith("geomcollection"): return "GeometryCollection" if db_datatype.startswith(("datetime", "timestamp")): return "DateTime" if db_datatype.startswith("date"): return "Date" if db_datatype.startswith("time"): return "Time" if db_datatype.startswith("year"): return "Year" if db_datatype.startswith("vector"): return "Vector" return "str" return "unknown" def get_enhanced_datatype_mapping(db_datatype, sdk_language): enhanced_map = { "Python": { "bool": "BoolField", "int": "IntField", "float": "FloatField", "str": "StringField", "Date": "DateField", "DateTime": "DateTimeField", "Time": "TimeField", "Year": "YearField", "Vector": "VectorField", } } if sdk_language == "TypeScript": # In TypeScript, the fields of type ${DatabaseObject} are the same as type ${DatabaseObject}Params return get_datatype_mapping(db_datatype, sdk_language) if sdk_language == "Python": # If `py_datatype` not in the list of types to be enhanced, # then we assume it shouldn't be enhanced and returned it as it is. py_datatype = get_datatype_mapping(db_datatype, sdk_language) return enhanced_map[sdk_language].get(py_datatype, py_datatype) return "unknown" def maybe_null(client_datatype, sdk_language): if sdk_language == "TypeScript": return f"MaybeNull<{client_datatype}>" if sdk_language == "Python": return f"Optional[{client_datatype}]" return "unknown" def get_procedure_datatype_mapping(sp_datatype, sdk_language): if sdk_language == "TypeScript": if sp_datatype == "BOOLEAN": return "boolean" if sp_datatype in ("NUMBER", "INT"): return "number" if sp_datatype == "JSON": return "object" return "string" return "unknown" def get_interface_datatype( field, sdk_language, class_name="", reference_class_name_postfix="", enhanced_fields=False, nullable=True, ): db_column_info = field.get("db_column") if db_column_info: db_datatype = db_column_info.get("datatype") db_not_null = db_column_info.get("not_null") # Todo: Handle SDK Options if not enhanced_fields: client_datatype = get_datatype_mapping(db_datatype, sdk_language) else: client_datatype = get_enhanced_datatype_mapping(db_datatype, sdk_language) if not nullable or db_not_null is True: return client_datatype return maybe_null(client_datatype, sdk_language) class_name_postfix = generate_identifier(value=field.get("name"), primitive="class", existing_identifiers=[]) return f"I{class_name}{reference_class_name_postfix}{class_name_postfix}" def datatype_is_primitive(client_datatype, sdk_language): # for now, consider only the data types we actually are able to map into if client_datatype is None: return False if sdk_language == "TypeScript": if client_datatype.startswith(("boolean", "number", "string")): return True return False if sdk_language == "Python": if client_datatype.startswith( ("bool", "float", "int", "str", "list", "tuple", "dict") ): return True return False return False def field_is_pk(field): if field.get("lev") == 1: db_column_info = field.get("db_column") if db_column_info and db_column_info.get("is_primary"): return True return False def field_is_unique(field): if field.get("lev") == 1: db_column_info = field.get("db_column") if db_column_info and (db_column_info.get("is_primary") or db_column_info.get("is_unique")): return True return False def field_is_nullable(field): if field.get("lev") != 1: return False db_column_info = field.get("db_column") if db_column_info is None: return False if db_column_info.get("not_null"): return False return True def field_has_row_ownership(field, obj): field_id = field.get("id") row_ownership_field_id = obj.get("row_ownership_field_id") return row_ownership_field_id is not None and field_id == row_ownership_field_id def field_is_required(field, obj): if field.get("lev") != 1: return False db_column_info = field.get("db_column") if db_column_info is None: return False # if a field can be NULL it is, by definition, optional not_null = db_column_info.get("not_null") # if a field is a primary key with AUTO_INCREMENt, it is optional id_generation = db_column_info.get("id_generation") # if a field has a default column value, it is optional column_default = db_column_info.get("column_default") # if a field is maps to row ownership column, it should also be optional has_row_ownership = field_has_row_ownership(field, obj) if not_null and id_generation is None and column_default is None and not has_row_ownership: return True return False def field_can_be_cursor(field): if field.get("lev") != 1: return False db_column_info = field.get("db_column") if db_column_info is None: return False # the column can be used as a cursor if it is a unique and sequential # e.g. if it is a timestamp or has an AUTO_INCREMENT constraint id_generation = db_column_info.get("id_generation") datatype = db_column_info.get("datatype") if ((id_generation is not None and id_generation.startswith("auto_inc")) or (datatype is not None and datatype.startswith("timestamp"))): return True return False def field_is_sortable(field): return field.get("allow_sorting", False) def get_field_by_id(fields, identifier): for field in fields: if field.get("id") == identifier: return field return None def get_reduced_field_interface_datatype(field, fields, sdk_language, class_name): if field.get("represents_reference_id"): obj_ref = field.get("object_reference") # Check if the field should be reduced to the value of another field ref_field_id = obj_ref.get("reduce_to_value_of_field_id") if obj_ref and ref_field_id: # Convert id to binary ref_field_id = lib.core.id_to_binary( ref_field_id, "reduce_to_value_of_field_id") # Lookup the field to reduce to ref_field = get_field_by_id(fields, ref_field_id) if ref_field: datatype = get_interface_datatype( ref_field, sdk_language, class_name) # If the reference mapping is "to_many", use an array ref_mapping = obj_ref.get("reference_mapping") is_array = "[]" if ref_mapping and ref_mapping.get( "to_many") == True else "" return datatype + is_array return None def generate_type_declaration( name, parents=[], fields={}, sdk_language="TypeScript", ignore_base_types=False, non_mandatory_fields: set[str] = set(), # Users may or not specify them requires_placeholder=False, is_unpacked=False, readonly_fields: set[str] = set(), ): if len(fields) == 0: if not requires_placeholder: return "" return generate_type_declaration_placeholder(name, sdk_language, is_unpacked) if sdk_language == "TypeScript": field_block = [ generate_type_declaration_field( name, value, sdk_language, non_mandatory=(name in non_mandatory_fields), allowed_special_characters={"(", ")"}, readonly=name in readonly_fields, ) for name, value in fields.items() ] if len(parents) > 0: # To avoid issues with optional fields, we always use type intersection to represent inheritance. intersection_block = f" & {' & '.join(parents)}" return ( f"export type I{name} = {{\n" + "".join(field_block) + f"}}{intersection_block};\n\n" ) # To follow the internal convention, we use interfaces for every other case. return f"export interface I{name} {{\n" + "".join(field_block) + "}\n\n" if sdk_language == "Python": field_block = [ generate_type_declaration_field( name, value, sdk_language, non_mandatory=( name in non_mandatory_fields and len(non_mandatory_fields) != len(fields) ), ) for name, value in fields.items() ] ordered_parents = ( [*parents] if ignore_base_types is True else ["TypedDict", *parents] ) if len(non_mandatory_fields) == len(fields): inheritance_block = f"({', '.join(ordered_parents)}, total=False)" else: inheritance_block = f"({', '.join(ordered_parents)})" return f"class I{name}{inheritance_block}:\n" + "".join(field_block) + "\n\n" def generate_type_declaration_field( name, value, sdk_language, non_mandatory=False, allowed_special_characters=None, readonly=False ): name = generate_identifier( value=name, sdk_language=sdk_language, existing_identifiers=[], allowed_special_characters=allowed_special_characters, ) indent = " " * 4 if sdk_language == "TypeScript": field_name_part = f"{indent}{"readonly " if readonly else ""}{name}?" if non_mandatory else f"{indent}{"readonly " if readonly else ""}{name}" if isinstance(value, list): return f"{field_name_part}: {value[0]}[],\n" return f"{field_name_part}: {value},\n" if sdk_language == "Python": if isinstance(value, list): hint = f"NotRequired[list[{value[0]}]]" if non_mandatory is True else f"list[{value[0]}]" else: hint = f"NotRequired[{value}]" if non_mandatory is True else f"{value}" return f'{indent}{name}: {hint}\n' def generate_data_class( name, fields, sdk_language, db_object_crud_ops: list[str], obj_endpoint: Optional[str] = None, primary_key_fields: set[str] = set(), ): if sdk_language == "TypeScript": if len(primary_key_fields) > 0: if "UPDATE" in db_object_crud_ops: fields.update({ "update()": f"Promise" }) if "DELETE" in db_object_crud_ops: fields.update({ "delete()": f"Promise" }) return generate_type_declaration( name=name, fields=fields, sdk_language=sdk_language, non_mandatory_fields=set(fields).difference({"update()", "delete()"}), readonly_fields=primary_key_fields ) if sdk_language == "Python": field_type_block = [ ( generate_type_declaration_field(name, value, sdk_language).rstrip() + " | UndefinedDataClassField\n" ) for name, value in fields.items() ] field_profile = [] for field_name, type_hint in [ ( generate_identifier(value=field, sdk_language=sdk_language, existing_identifiers=[]), value[0] if isinstance(value, list) else value, ) for field, value in fields.items() ]: field_profile.append(f'"{field_name}": {type_hint}') join_field_profile = ( "{\n" + f"{" "*16}" + f",\n{" "*16}".join(field_profile).rstrip() + f"\n{" "*12}" + "}" ) mixins = [] if len(primary_key_fields) > 0: if "UPDATE" in db_object_crud_ops: mixins.append(f'\n\t_MrsDocumentUpdateMixin["I{name}Data", "I{name}", "I{name}Details"],') if "DELETE" in db_object_crud_ops: mixins.append(f'\n\t_MrsDocumentDeleteMixin["I{name}Data", "I{name}Filterable"],') if mixins: mixins.append("\n\t") return SDK_PYTHON_DATACLASS_TEMPLATE.format( name=name, join_field_block="".join(field_type_block).rstrip(), obj_endpoint=obj_endpoint, field_profile=join_field_profile, primary_key_name=( None if len(primary_key_fields) == 0 else f'"{",".join(primary_key_fields)}"' ), mixins="".join(mixins), ) def generate_field_enum(name, fields=None, sdk_language="TypeScript"): if sdk_language == "TypeScript": # In TypeScript the field enum can be obtained using keyof on the original object type declaration, so there is # no need for an additional type declaration. return "" if sdk_language == "Python": if not fields or len(fields) == 0: # To avoid conditional logic in the template, we can generate a placeholder declaration regardless. return generate_type_declaration_placeholder(f"{name}Field", sdk_language) fields_in_case = [lib.core.convert_to_snake_case(field) for field in fields] return generate_enum(f"{name}Field", fields_in_case, sdk_language) def generate_enum(name, values, sdk_language): enum_def = generate_literal_type(values, sdk_language) if sdk_language == "TypeScript": return f"export type I{name} = {enum_def};\n\n" if sdk_language == "Python": return f"I{name}: TypeAlias = {enum_def}\n\n\n" def generate_type_declaration_placeholder(name, sdk_language, is_unpacked=False): if sdk_language == "TypeScript": return f"type I{name} = never;\n\n" if sdk_language == "Python": if not is_unpacked: return f"I{name}: TypeAlias = None\n\n\n" # Using an empty TypedDict helps to avoid issues with Unpack return f"class I{name}(TypedDict):\n pass\n\n\n" def generate_literal_type(values, sdk_language): if sdk_language == "TypeScript": items = [f'"{value}"' for value in values] return f"{' | '.join(items)}" if sdk_language == "Python": items = [f'{" " * 4}"{value}"' for value in values] s = ',\n'.join(items) return (f"Literal[\n" + f"{s},\n" + "]") def generate_selectable(name, fields, sdk_language): if sdk_language == "TypeScript": return "" if sdk_language == "Python": return generate_type_declaration(name=f"{name}Selectable", fields={ field: "bool" for field in fields }, sdk_language=sdk_language, non_mandatory_fields=set(fields)) def generate_sortable( name: str, fields: dict[str, str] = {}, sdk_language: Literal["TypeScript", "Python"] = "TypeScript", ): if sdk_language == "TypeScript": return "" if sdk_language == "Python": return generate_type_declaration( name=f"{name}Sortable", fields={field: "Order" for field in fields}, sdk_language=sdk_language, non_mandatory_fields=set(fields), requires_placeholder=True, ) def generate_union(name, types, sdk_language): if sdk_language == "TypeScript": return f"export type {name} = {' | '.join(types)};\n\n" if sdk_language == "Python": return f"{name}: TypeAlias = {' | '.join(types)}\n\n\n" def generate_sequence_constant(name, values, sdk_language): if sdk_language == "TypeScript": return f"const {name} = {json.dumps(values)} as const;\n" elif sdk_language == "Python": return f"{name}: Sequence = {json.dumps(values)}\n\n" def object_is_routine(db_obj, of_type: set[str] = {"PROCEDURE", "FUNCTION", "SCRIPT"}): return db_obj.get("object_type") in of_type def generate_interfaces( db_obj, obj, fields, class_name, sdk_language, db_object_crud_ops: list[str], obj_endpoint: Optional[str] = None, ): obj_interfaces: list[str] = [] interface_fields = {} param_interface_fields = {} out_params_interface_fields = {} obj_unique_fields = {} obj_cursor_fields = {} obj_sortable_fields: set[str] = set() has_nested_fields = False required_datatypes: set[str] = set() # The I{class_name}, I{class_name}Params and I{class_name}Out interfaces for field in fields: db_column = field.get("db_column", {}) # The field needs to be on level 1 and enabled if field.get("lev") == 1 and field.get("enabled"): datatype = get_interface_datatype(field, sdk_language, class_name) enhanced_fields = False if object_is_routine(db_obj) else True enhanced_datatype = get_interface_datatype( field=field, sdk_language=sdk_language, class_name=class_name, enhanced_fields=enhanced_fields, ) # Handle references if field.get("represents_reference_id"): has_nested_fields = True # Check if the field should be reduced to the value of another field reduced_to_datatype = get_reduced_field_interface_datatype( field, fields, sdk_language, class_name ) if reduced_to_datatype: interface_fields.update({field.get("name"): reduced_to_datatype}) else: obj_ref = field.get("object_reference") # Add field if the referred table is not unnested if not obj_ref.get("unnest"): # If this field represents an OUT parameter of a SP, add it to the # out_params_interface_fields list if obj.get("kind") == "PARAMETERS" and object_is_routine( db_obj ): if db_column.get("in"): param_interface_fields.update( {field.get("name"): datatype} ) if db_column.get("out"): out_params_interface_fields.update( {field.get("name"): datatype} ) elif obj.get("kind") == "PARAMETERS" and not object_is_routine( db_obj ): param_interface_fields.update( {field.get("name"): enhanced_datatype} ) elif field.get("allow_filtering") and not object_is_routine( db_obj ): # RESULT interface_fields.update({field.get("name"): datatype}) # Add all table fields that have allow_filtering set and SP params to the # param_interface_fields param_interface_fields.update( {field.get("name"): enhanced_datatype} ) # Call recursive interface generation generate_nested_interfaces( obj_interfaces, interface_fields, field, reference_class_name_postfix=lib.core.convert_path_to_pascal_case( field.get("name") ), fields=fields, class_name=class_name, sdk_language=sdk_language, ) elif obj.get("kind") == "PARAMETERS": # If this field represents an OUT parameter of a SP, add it to the # out_params_interface_fields list if db_column.get("in"): param_interface_fields.update({field.get("name"): datatype}) if db_column.get("out"): out_params_interface_fields.update({field.get("name"): datatype}) else: interface_fields.update({field.get("name"): datatype}) # Add all table fields that have allow_filtering set and SP params to the param_interface_fields if field.get("allow_filtering") and not object_is_routine(db_obj): param_interface_fields.update( {field.get("name"): enhanced_datatype} ) if not object_is_routine(db_obj): enhanced_datatype = get_interface_datatype( field=field, sdk_language=sdk_language, class_name=class_name, enhanced_fields=True, nullable=False, ) # Build Unique list if field_is_unique(field): obj_unique_fields.update({field.get("name"): enhanced_datatype}) # Build list of columns which can potentially be used for cursor-based pagination. if field_can_be_cursor(field): obj_cursor_fields.update({field.get("name"): enhanced_datatype}) # Build list of sortable fields if field_is_sortable(field): obj_sortable_fields.add(field.get("name")) if not object_is_routine(db_obj): # The object is a TABLE or a VIEW if sdk_language != "TypeScript": # These type declarations are not needed for TypeScript because it uses a Proxy to replace the interface # and not a wrapper class. This might change in the future. mrs_resource_type = "IMrsResourceDetails" required_datatypes.add(mrs_resource_type) obj_interfaces.append( generate_type_declaration( name=f"{class_name}Details", parents=[mrs_resource_type], fields=interface_fields, sdk_language=sdk_language, ignore_base_types=True, non_mandatory_fields=set(interface_fields), ) ) obj_interfaces.append( generate_type_declaration( name=f"{class_name}Data", fields=interface_fields, sdk_language=sdk_language, non_mandatory_fields=set(interface_fields), ) ) if "CREATE" in db_object_crud_ops: obj_non_mandatory_fields = set( [ field.get("name") for field in fields # exclude fields that are out of range (e.g. on different nesting levels) if field.get("name") in interface_fields.keys() and field_is_required(field, obj) is False ] ) obj_interfaces.append( generate_type_declaration( name=f"New{class_name}", fields=interface_fields, sdk_language=sdk_language, non_mandatory_fields=obj_non_mandatory_fields, ) ) if "UPDATE" in db_object_crud_ops: # TODO: No partial update is supported yet. Once it is, the # `non-mandatory_fields` argument should not change. # This way, users can know what fields are required and which ones aren't. # However, even when replacing an entire resource, it should be possible to # unset nullable fields. nullable_fields = [ field.get("name") for field in fields if field_is_nullable(field) or field_has_row_ownership(field, obj) ] obj_interfaces.append( generate_type_declaration( name=f"Update{class_name}", fields=interface_fields, sdk_language=sdk_language, non_mandatory_fields=set(nullable_fields), ) ) primary_key_fields = [field.get("name") for field in fields if field_is_pk(field)] obj_interfaces.append( generate_data_class( name=class_name, fields=interface_fields, sdk_language=sdk_language, db_object_crud_ops=db_object_crud_ops, obj_endpoint=obj_endpoint, primary_key_fields=set(primary_key_fields), ) ) obj_interfaces.append( generate_field_enum( name=class_name, fields=interface_fields, sdk_language=sdk_language ) ) if not has_nested_fields: # This creates a type alias for something like None, which ensures there is always a default # value for *{class_name}NestedField and saves us from using conditionals in the template. obj_interfaces.append( generate_field_enum( name=f"{class_name}Nested", sdk_language=sdk_language ) ) obj_interfaces.append( generate_selectable(class_name, interface_fields, sdk_language) ) obj_interfaces.append( generate_sortable(class_name, obj_sortable_fields, sdk_language) ) construct_parents = ( [] if sdk_language != "Python" else ["Generic[Filterable]", "HighOrderOperator[Filterable]"] ) obj_interfaces.append( generate_type_declaration( name=f"{class_name}Filterable", parents=construct_parents, fields=param_interface_fields, sdk_language=sdk_language, ignore_base_types=True, non_mandatory_fields=set(param_interface_fields), ) ) obj_interfaces.append( generate_type_declaration( name=f"{class_name}UniqueFilterable", fields=obj_unique_fields, sdk_language=sdk_language, non_mandatory_fields=set(obj_unique_fields), ) ) obj_interfaces.append( generate_type_declaration( name=f"{class_name}Cursors", fields=obj_cursor_fields, sdk_language=sdk_language, non_mandatory_fields=set(obj_cursor_fields), # To avoid conditional logic in the template, we should generate a void type declaration. requires_placeholder=True, ) ) # FUNCTIONs, PROCEDUREs and SCRIPTs elif obj.get("kind") == "RESULT": obj_interfaces.append( generate_type_declaration( name=class_name, fields=interface_fields, sdk_language=sdk_language, non_mandatory_fields=set(interface_fields), ) ) if len(interface_fields) > 0: # Result sets are non-deterministic and there is no way to know if a column value can be NULL. # Thus, we must assume that is always the case. if sdk_language == "TypeScript": required_datatypes.add("MaybeNull") result_fields = { "type": generate_literal_type([class_name], sdk_language), "items": [f"I{class_name}"], } obj_interfaces.append( generate_type_declaration( name=f"Tagged{class_name}", fields=result_fields, sdk_language=sdk_language, parents=( ["JsonObject"] if sdk_language == "TypeScript" else [] ), # TypeScript tagged unions inherit from JsonObject ) ) # The Python SDK uses a generic dataclass for the tagged union of result set types. # The generic dataclass needs to know the possible values of the "type" field. if sdk_language == "Python": obj_interfaces.append( generate_enum( name=f"{class_name}Type", values=[class_name], sdk_language=sdk_language, ) ) else: # kind = "PARAMETERS" if len(param_interface_fields) > 0: # Parameters are "optional" in a way that they can be NULL at the SQL level. if sdk_language == "TypeScript": required_datatypes.add("MaybeNull") # Type definition for the set of IN/INOUT Parameters. obj_interfaces.append( generate_type_declaration( name=f"{class_name}Params", fields=param_interface_fields, sdk_language=sdk_language, non_mandatory_fields=set(param_interface_fields), # To avoid conditional logic in the template, we should generate a void type declaration. requires_placeholder=True, is_unpacked=True, ) ) # Type definition for the set of OUT/INOUT Parameters. obj_interfaces.append( generate_type_declaration( name=f"{class_name}ParamsOut", fields=out_params_interface_fields, sdk_language=sdk_language, non_mandatory_fields=set(out_params_interface_fields), # To avoid conditional logic in the template, we should generate a void type declaration. # In this case, the placeholder is only needed for Procedures, because the type declaration # is not used otherwise. requires_placeholder=object_is_routine(db_obj, of_type={"PROCEDURE"}), ) ) return "".join(obj_interfaces), required_datatypes # For now, this function is not used for ${DatabaseObject}Params type declarations def generate_nested_interfaces( obj_interfaces, parent_interface_fields, parent_field, reference_class_name_postfix, fields, class_name, sdk_language): # Build interface name interface_name = f"{class_name}{reference_class_name_postfix}" # Check if the reference has unnest set, and if so, use the parent_interface_fields parent_obj_ref = parent_field.get("object_reference") interface_fields = {} if not parent_obj_ref.get( "unnest") else parent_interface_fields for field in fields: if (field.get("parent_reference_id") == parent_field.get("represents_reference_id") and field.get("enabled")): # Handle references if field.get("represents_reference_id"): # Check if the field should be reduced to the value of another field reduced_to_datatype = get_reduced_field_interface_datatype( field, fields, sdk_language, class_name) if reduced_to_datatype: interface_fields.update({ field.get("name"): reduced_to_datatype }) else: obj_ref = field.get("object_reference") field_interface_name = lib.core.convert_path_to_pascal_case( field.get("name")) # Add field if the referred table is not unnested if not obj_ref.get("unnest"): datatype = f"{class_name}{reference_class_name_postfix + field.get("name")}" # Should use the corresponding nested field type. interface_fields.update({ field.get("name"): interface_name + field_interface_name }) # If not, do recursive call generate_nested_interfaces( obj_interfaces, interface_fields, field, reference_class_name_postfix=reference_class_name_postfix + field_interface_name, fields=fields, class_name=class_name, sdk_language=sdk_language) else: datatype = get_interface_datatype(field, sdk_language) interface_fields.update({ field.get("name"): datatype }) if not parent_obj_ref.get("unnest"): obj_interfaces.append(generate_type_declaration(name=interface_name, fields=interface_fields, sdk_language=sdk_language)) obj_interfaces.append(generate_field_enum(name=f"{class_name}Nested", fields=interface_fields, sdk_language=sdk_language)) def get_mrs_runtime_management_code(session, service_url=None): """Returns TypeScript code that allows the user to manage specific MRS runtime settings. This includes the configuration status of MRS with an info string giving context sensitive information how and the list of available REST services with their URL and and methods to set which REST services is the current one and to print the SDK runtime code. Args: session (object): The database session to use. service_url (str): The URL of the current REST service Returns: The TypeScript code that defines and sets the mrs object """ status = lib.general.get_status(session) if status['service_configured'] is True: services = lib.services.get_services(session=session) else: services = [] s = """ class MrsService { #serviceId: string; public constructor(serviceId: string) { this.#serviceId = serviceId; } public setAsCurrent = () => { mrsSetCurrentService(this.#serviceId); }; public edit = () => { mrsEditService(this.#serviceId); }; public exportSdk = () => { mrsExportServiceSdk(this.#serviceId); }; } class Mrs { """ service_list = [] for service in services: service_name = lib.core.convert_path_to_camel_case( service.get("url_context_root")) service_id = lib.core.convert_id_to_string(service.get("id")) # If no explicit service_url is given, use the service's host_ctx and url_context_root if not service_url or service.get("is_current") == 0: host_ctx = service.get("host_ctx") url_context_root = service.get("url_context_root") # If no host_ctx starting with http is given, default to https://localhost:8443 if not host_ctx.lower().startswith("http"): service_url = "https://localhost:8443" + url_context_root else: service_url = url_context_root + url_context_root if service.get("is_current") == 1: s += f' public {service_name} = {service_name};\n' else: s += f' public {service_name}: MrsService = new MrsService("{service_id}");\n' service_list.append({ "serviceName": service_name, "url": service_url, "isCurrent": service.get("is_current") == 1}) if status['service_configured'] is False: status_output = { "configured": False, "info": "The MySQL REST Service has not been configured on this MySQL instance yet. Switch to " + "SQL mode and use the CONFIGURE REST METADATA command to configure the instance.", "services": []} elif len(service_list) == 0: status_output = { "configured": True, "info": "No REST service has been created yet. Switch to SQL Mode and use the " + "CREATE REST SERVICE command to create a new REST service.", "services": []} else: status_output = { "configured": True, "info": f'{len(service_list)} REST service{"s" if len(service_list) > 1 else ""} available.', "services": service_list} s += f""" public getStatus = () => {{ return {json.dumps(status_output)}; }} public addService = () => {{ mrsEditService(); }} public printSdkCode = () => {{ mrsPrintSdkCode(); }} public refreshSdkCode = () => {{ mrsRefreshSdkCode(); }} }} const mrs = new Mrs(); """ return s def replace_group_1_match_only(m): return "" if m.group(1) is not None else m.group(0) def remove_js_whitespace_and_comments(code): # Remove comments. This regex does not match single line comments that # include a " character, which is ignored for simplicity. code = re.sub(r"/\*.*?\*/|//[^\"]*?$", "", code, flags=re.DOTALL | re.MULTILINE) # Remove all empty linebreaks code = re.sub(r"^\s*\n", "", code, flags=re.DOTALL | re.MULTILINE) # Substitute leading spaces with tabs code = re.sub(r" ", "\t", code, flags=re.DOTALL | re.MULTILINE) return code