mrs_plugin/schemas.py
def change_schema(**kwargs):
    """Makes a given change to a MRS service

    Args:
        **kwargs: Additional options

    Keyword Args:
        change_type (int): Type of change
        schema_name (str): The name of the schema
        request_path (str): The request_path of the schema
        service_id (int): The id of the service
        value (str): The values as string or dict
        schema_id (int): The id of the schema
        session (object): The database session to use
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
        The result message as string
    """
    import json

    change_type = kwargs.get("change_type")
    schema_name = kwargs.get("schema_name")
    request_path = kwargs.get("request_path")
    service_id = kwargs.get("service_id")
    value = kwargs.get("value")

    schema_id = kwargs.get("schema_id")
    session = kwargs.get("session")
    interactive = kwargs.get("interactive", core.get_interactive_default())
    raise_exceptions = kwargs.get("raise_exceptions", not interactive)

    try:
        session = core.get_current_session(session)

        # Make sure the MRS metadata schema exists and has the right version
        core.ensure_rds_metadata_schema(session)

        # The list of schemas to be changed
        schema_ids = [schema_id] if schema_id else []
        include_enable_state = None

        if not schema_id:
            # Check if given service_id exists or use the current service
            service = mrs_services.get_service(
                service_id=service_id, session=session, interactive=interactive,
                return_formatted=False)

            if not schema_name and not request_path and interactive:
                schemas = get_schemas(
                    service_id=service.get("id"),
                    include_enable_state=include_enable_state,
                    session=session, interactive=False)
                caption = ("Please select a schema index, type "
                           "the request_path or type '*' "
                           "to select all: ")
                selection = core.prompt_for_list_item(
                    item_list=schemas,
                    prompt_caption=caption,
                    item_name_property="request_path",
                    given_value=None,
                    print_list=True,
                    allow_multi_select=True)
                if not selection:
                    raise ValueError("Operation cancelled.")

                schema_ids = [item["id"] for item in selection]

            if schema_name:
                # Lookup the schema name
                res = session.run_sql(
                    """
                    SELECT id FROM `mysql_rest_service_metadata`.db_schema
                    WHERE name = ? AND service_id = ?
                    """,
                    [schema_name, service.get("id")])
                rows = res.fetch_all()
                if rows:
                    for row in rows:
                        schema_ids.append(row.get_field("id"))

            if request_path:
                # Lookup the schema name
                res = session.run_sql(
                    """
                    SELECT id FROM `mysql_rest_service_metadata`.db_schema
                    WHERE request_path = ? AND service_id = ?
                    """,
                    [request_path, service.get("id")])
                row = res.fetch_one()
                if row:
                    schema_ids.append(row.get_field("id"))

        if len(schema_ids) == 0:
            raise ValueError("The specified schema was not found.")

        # Check the given value
        if interactive and not value:
            if change_type == SCHEMA_SET_REQUIRES_AUTH:
                value = prompt_for_requires_auth()
            elif change_type == SCHEMA_SET_ITEMS_PER_PAGE:
                value = prompt_for_items_per_page()
            elif change_type == SCHEMA_SET_COMMENTS:
                value = core.prompt_for_comments()

        # Update all given services
        for schema_id in schema_ids:
            params = [schema_id]

            # The REQUEST_PATH has to be checked for each schema
            if change_type == SCHEMA_SET_REQUEST_PATH:
                request_path_val = value
            elif change_type == SCHEMA_SET_ALL:
                if type(value) == str:  # TODO: Check why dicts cannot be used
                    value = json.loads(value)
                request_path_val = value.get("request_path")

            if (change_type == SCHEMA_SET_REQUEST_PATH or
               change_type == SCHEMA_SET_ALL):
                schema = get_schema(
                    request_path=request_path,
                    schema_id=schema_id, session=session,
                    interactive=False, return_formatted=False)

                if interactive and not request_path:
                    request_path_val = prompt_for_request_path(
                        schema.get("name"))

                # If the request path has changed, check if the new one is valid
                if request_path_val != schema.get("request_path"):
                    if (not request_path_val or
                            not request_path_val.startswith('/')):
                        raise ValueError(
                            "The url_context_root has to start with '/'.")

                    core.check_request_path(
                        schema.get("host_ctx") + request_path_val, session=session)

            if change_type == SCHEMA_DISABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET enabled = FALSE
                    WHERE id = ?
                    """
            elif change_type == SCHEMA_ENABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET enabled = TRUE
                    WHERE id = ?
                    """
            elif change_type == SCHEMA_DELETE:
                session.run_sql("""
                    DELETE FROM `mysql_rest_service_metadata`.db_object
                    WHERE db_schema_id = ?
                    """, [schema_id])
                sql = """
                    DELETE FROM `mysql_rest_service_metadata`.db_schema
                    WHERE id = ?
                    """
            elif change_type == SCHEMA_SET_NAME:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET name = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_REQUEST_PATH:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET request_path = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_REQUIRES_AUTH:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET requires_auth = ?
                    WHERE id = ?
                    """
                params.insert(0, str(value).lower() == "true")
            elif change_type == SCHEMA_SET_ITEMS_PER_PAGE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET items_per_page = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_COMMENTS:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET comments = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_ALL:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.`db_schema`
                    SET name = ?,
                        request_path = ?,
                        requires_auth = ?,
                        enabled = ?,
                        items_per_page = ?,
                        comments = ?
                    WHERE id = ?
                    """
                params.insert(0, value.get("name", ""))
                params.insert(1, request_path_val)
                params.insert(
                    2, (str(value.get("requires_auth")).lower() == "true" or
                    str(value.get("requires_auth")) == "1"))
                params.insert(
                    3, (str(value.get("enabled")).lower() == "true" or
                    str(value.get("enabled")) == "1"))
                params.insert(4, value.get("items_per_page", 25))
                params.insert(5, value.get("comments", ""))
            else:
                raise Exception("Operation not supported")

            res = session.run_sql(sql, params)
            if res.get_affected_row_count() == 0:
                raise Exception(
                    f"The specified schema with id {schema_id} was not "
                    "found.")

        if len(schema_ids) == 1:
            msg = "The schema has been "
        else:
            msg = "The schemas have been "

        if change_type == SCHEMA_DISABLE:
            msg += "disabled."
        elif change_type == SCHEMA_ENABLE:
            msg += "enabled."
        elif change_type == SCHEMA_DELETE:
            msg += "deleted."
        else:
            msg += "updated."

        return msg

    except Exception as e:
        if raise_exceptions:
            raise
        print(f"Error: {str(e)}")
Makes a given change to a MRS service

    Args:
        **kwargs: Additional options

    Keyword Args:
        change_type (int): Type of change
        schema_name (str): The name of the schema
        request_path (str): The request_path of the schema
        service_id (int): The id of the service
        value (str): The values as string or dict
        schema_id (int): The id of the schema
        session (object): The database session to use
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
        The result message as string
    
Makes a given change to a MRS service
Makes a given change to a MRS service
def change_schema(**kwargs):
    import json
    change_type = kwargs.get("change_type")
    schema_name = kwargs.get("schema_name")
    request_path = kwargs.get("request_path")
    service_id = kwargs.get("service_id")
    value = kwargs.get("value")
    schema_id = kwargs.get("schema_id")
    session = kwargs.get("session")
    interactive = kwargs.get("interactive", core.get_interactive_default())
    raise_exceptions = kwargs.get("raise_exceptions", not interactive)
    try:
        session = core.get_current_session(session)
        core.ensure_rds_metadata_schema(session)
        schema_ids = [schema_id] if schema_id else []
        include_enable_state = None
        if not schema_id:
            service = mrs_services.get_service(
                service_id=service_id, session=session, interactive=interactive,
                return_formatted=False)
            if not schema_name and not request_path and interactive:
                schemas = get_schemas(
                    service_id=service.get("id"),
                    include_enable_state=include_enable_state,
                    session=session, interactive=False)
                caption = ("Please select a schema index, type "
                           "the request_path or type '*' "
                           "to select all: ")
                selection = core.prompt_for_list_item(
                    item_list=schemas,
                    prompt_caption=caption,
                    item_name_property="request_path",
                    given_value=None,
                    print_list=True,
                    allow_multi_select=True)
                if not selection:
                    raise ValueError("Operation cancelled.")
                schema_ids = [item["id"] for item in selection]
            if schema_name:
                res = session.run_sql(
                    """
                    SELECT id FROM `mysql_rest_service_metadata`.db_schema
                    WHERE name = ? AND service_id = ?
                    """,
                    [schema_name, service.get("id")])
                rows = res.fetch_all()
                if rows:
                    for row in rows:
                        schema_ids.append(row.get_field("id"))
            if request_path:
                res = session.run_sql(
                    """
                    SELECT id FROM `mysql_rest_service_metadata`.db_schema
                    WHERE request_path = ? AND service_id = ?
                    """,
                    [request_path, service.get("id")])
                row = res.fetch_one()
                if row:
                    schema_ids.append(row.get_field("id"))
        if len(schema_ids) == 0:
            raise ValueError("The specified schema was not found.")
        if interactive and not value:
            if change_type == SCHEMA_SET_REQUIRES_AUTH:
                value = prompt_for_requires_auth()
            elif change_type == SCHEMA_SET_ITEMS_PER_PAGE:
                value = prompt_for_items_per_page()
            elif change_type == SCHEMA_SET_COMMENTS:
                value = core.prompt_for_comments()
        for schema_id in schema_ids:
            params = [schema_id]
            if change_type == SCHEMA_SET_REQUEST_PATH:
                request_path_val = value
            elif change_type == SCHEMA_SET_ALL:
                if type(value) == str:  
                    value = json.loads(value)
                request_path_val = value.get("request_path")
            if (change_type == SCHEMA_SET_REQUEST_PATH or
               change_type == SCHEMA_SET_ALL):
                schema = get_schema(
                    request_path=request_path,
                    schema_id=schema_id, session=session,
                    interactive=False, return_formatted=False)
                if interactive and not request_path:
                    request_path_val = prompt_for_request_path(
                        schema.get("name"))
                if request_path_val != schema.get("request_path"):
                    if (not request_path_val or
                            not request_path_val.startswith('/')):
                        raise ValueError(
                            "The url_context_root has to start with '/'.")
                    core.check_request_path(
                        schema.get("host_ctx") + request_path_val, session=session)
            if change_type == SCHEMA_DISABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET enabled = FALSE
                    WHERE id = ?
                    """
            elif change_type == SCHEMA_ENABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET enabled = TRUE
                    WHERE id = ?
                    """
            elif change_type == SCHEMA_DELETE:
                session.run_sql("""
                    DELETE FROM `mysql_rest_service_metadata`.db_object
                    WHERE db_schema_id = ?
                    """, [schema_id])
                sql = """
                    DELETE FROM `mysql_rest_service_metadata`.db_schema
                    WHERE id = ?
                    """
            elif change_type == SCHEMA_SET_NAME:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET name = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_REQUEST_PATH:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET request_path = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_REQUIRES_AUTH:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET requires_auth = ?
                    WHERE id = ?
                    """
                params.insert(0, str(value).lower() == "true")
            elif change_type == SCHEMA_SET_ITEMS_PER_PAGE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET items_per_page = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_COMMENTS:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET comments = ?
                    WHERE id = ?
                    """
                params.insert(0, value)
            elif change_type == SCHEMA_SET_ALL:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.`db_schema`
                    SET name = ?,
                        request_path = ?,
                        requires_auth = ?,
                        enabled = ?,
                        items_per_page = ?,
                        comments = ?
                    WHERE id = ?
                    """
                params.insert(0, value.get("name", ""))
                params.insert(1, request_path_val)
                params.insert(
                    2, (str(value.get("requires_auth")).lower() == "true" or
                    str(value.get("requires_auth")) == "1"))
                params.insert(
                    3, (str(value.get("enabled")).lower() == "true" or
                    str(value.get("enabled")) == "1"))
                params.insert(4, value.get("items_per_page", 25))
                params.insert(5, value.get("comments", ""))
            else:
                raise Exception("Operation not supported")
            res = session.run_sql(sql, params)
            if res.get_affected_row_count() == 0:
                raise Exception(
                    f"The specified schema with id {schema_id} was not "
                    "found.")
        if len(schema_ids) == 1:
            msg = "The schema has been "
        else:
            msg = "The schemas have been "
        if change_type == SCHEMA_DISABLE:
            msg += "disabled."
        elif change_type == SCHEMA_ENABLE:
            msg += "enabled."
        elif change_type == SCHEMA_DELETE:
            msg += "deleted."
        else:
            msg += "updated."
        return msg
    except Exception as e:
        if raise_exceptions:
            raise
        print(f"Error: {str(e)}")
def change_schema ( ** kwargs ) : import json change_type = kwargs . get ( "change_type" ) schema_name = kwargs . get ( "schema_name" ) request_path = kwargs . get ( "request_path" ) service_id = kwargs . get ( "service_id" ) value = kwargs . get ( "value" ) schema_id = kwargs . get ( "schema_id" ) session = kwargs . get ( "session" ) interactive = kwargs . get ( "interactive" , core . get_interactive_default ( ) ) raise_exceptions = kwargs . get ( "raise_exceptions" , not interactive ) try : session = core . get_current_session ( session ) core . ensure_rds_metadata_schema ( session ) schema_ids = [ schema_id ] if schema_id else [ ] include_enable_state = None if not schema_id : service = mrs_services . get_service ( service_id = service_id , session = session , interactive = interactive , return_formatted = False ) if not schema_name and not request_path and interactive : schemas = get_schemas ( service_id = service . get ( "id" ) , include_enable_state = include_enable_state , session = session , interactive = False ) caption = ( "Please select a schema index, type " "the request_path or type '*' " "to select all: " ) selection = core . prompt_for_list_item ( item_list = schemas , prompt_caption = caption , item_name_property = "request_path" , given_value = None , print_list = True , allow_multi_select = True ) if not selection : raise ValueError ( "Operation cancelled." ) schema_ids = [ item [ "id" ] for item in selection ] if schema_name : res = session . run_sql ( """
                    SELECT id FROM `mysql_rest_service_metadata`.db_schema
                    WHERE name = ? AND service_id = ?
                    """ , [ schema_name , service . get ( "id" ) ] ) rows = res . fetch_all ( ) if rows : for row in rows : schema_ids . append ( row . get_field ( "id" ) ) if request_path : res = session . run_sql ( """
                    SELECT id FROM `mysql_rest_service_metadata`.db_schema
                    WHERE request_path = ? AND service_id = ?
                    """ , [ request_path , service . get ( "id" ) ] ) row = res . fetch_one ( ) if row : schema_ids . append ( row . get_field ( "id" ) ) if len ( schema_ids ) == 0 : raise ValueError ( "The specified schema was not found." ) if interactive and not value : if change_type == SCHEMA_SET_REQUIRES_AUTH : value = prompt_for_requires_auth ( ) elif change_type == SCHEMA_SET_ITEMS_PER_PAGE : value = prompt_for_items_per_page ( ) elif change_type == SCHEMA_SET_COMMENTS : value = core . prompt_for_comments ( ) for schema_id in schema_ids : params = [ schema_id ] if change_type == SCHEMA_SET_REQUEST_PATH : request_path_val = value elif change_type == SCHEMA_SET_ALL : if type ( value ) == str : value = json . loads ( value ) request_path_val = value . get ( "request_path" ) if ( change_type == SCHEMA_SET_REQUEST_PATH or change_type == SCHEMA_SET_ALL ) : schema = get_schema ( request_path = request_path , schema_id = schema_id , session = session , interactive = False , return_formatted = False ) if interactive and not request_path : request_path_val = prompt_for_request_path ( schema . get ( "name" ) ) if request_path_val != schema . get ( "request_path" ) : if ( not request_path_val or not request_path_val . startswith ( '/' ) ) : raise ValueError ( "The url_context_root has to start with '/'." ) core . check_request_path ( schema . get ( "host_ctx" ) + request_path_val , session = session ) if change_type == SCHEMA_DISABLE : sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET enabled = FALSE
                    WHERE id = ?
                    """ elif change_type == SCHEMA_ENABLE : sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET enabled = TRUE
                    WHERE id = ?
                    """ elif change_type == SCHEMA_DELETE : session . run_sql ( """
                    DELETE FROM `mysql_rest_service_metadata`.db_object
                    WHERE db_schema_id = ?
                    """ , [ schema_id ] ) sql = """
                    DELETE FROM `mysql_rest_service_metadata`.db_schema
                    WHERE id = ?
                    """ elif change_type == SCHEMA_SET_NAME : sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET name = ?
                    WHERE id = ?
                    """ params . insert ( 0 , value ) elif change_type == SCHEMA_SET_REQUEST_PATH : sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET request_path = ?
                    WHERE id = ?
                    """ params . insert ( 0 , value ) elif change_type == SCHEMA_SET_REQUIRES_AUTH : sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET requires_auth = ?
                    WHERE id = ?
                    """ params . insert ( 0 , str ( value ) . lower ( ) == "true" ) elif change_type == SCHEMA_SET_ITEMS_PER_PAGE : sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET items_per_page = ?
                    WHERE id = ?
                    """ params . insert ( 0 , value ) elif change_type == SCHEMA_SET_COMMENTS : sql = """
                    UPDATE `mysql_rest_service_metadata`.db_schema
                    SET comments = ?
                    WHERE id = ?
                    """ params . insert ( 0 , value ) elif change_type == SCHEMA_SET_ALL : sql = """
                    UPDATE `mysql_rest_service_metadata`.`db_schema`
                    SET name = ?,
                        request_path = ?,
                        requires_auth = ?,
                        enabled = ?,
                        items_per_page = ?,
                        comments = ?
                    WHERE id = ?
                    """ params . insert ( 0 , value . get ( "name" , "" ) ) params . insert ( 1 , request_path_val ) params . insert ( 2 , ( str ( value . get ( "requires_auth" ) ) . lower ( ) == "true" or str ( value . get ( "requires_auth" ) ) == "1" ) ) params . insert ( 3 , ( str ( value . get ( "enabled" ) ) . lower ( ) == "true" or str ( value . get ( "enabled" ) ) == "1" ) ) params . insert ( 4 , value . get ( "items_per_page" , 25 ) ) params . insert ( 5 , value . get ( "comments" , "" ) ) else : raise Exception ( "Operation not supported" ) res = session . run_sql ( sql , params ) if res . get_affected_row_count ( ) == 0 : raise Exception ( f"The specified schema with id {schema_id} was not " "found." ) if len ( schema_ids ) == 1 : msg = "The schema has been " else : msg = "The schemas have been " if change_type == SCHEMA_DISABLE : msg += "disabled." elif change_type == SCHEMA_ENABLE : msg += "enabled." elif change_type == SCHEMA_DELETE : msg += "deleted." else : msg += "updated." return msg except Exception as e : if raise_exceptions : raise print ( f"Error: {str(e)}" )
Makes a given change to a MRS service
Makes a given change to a MRS service
"""Makes a given change to a MRS service

    Args:
        **kwargs: Additional options

    Keyword Args:
        change_type (int): Type of change
        schema_name (str): The name of the schema
        request_path (str): The request_path of the schema
        service_id (int): The id of the service
        value (str): The values as string or dict
        schema_id (int): The id of the schema
        session (object): The database session to use
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
        The result message as string
    """ # Make sure the MRS metadata schema exists and has the right version # The list of schemas to be changed # Check if given service_id exists or use the current service # Lookup the schema name # Lookup the schema name # Check the given value # Update all given services # The REQUEST_PATH has to be checked for each schema # TODO: Check why dicts cannot be used # If the request path has changed, check if the new one is valid
The result message as string
The result message as string
**kwargs
None