mrs_plugin/db_objects.py
def change_db_object(change_type=None, db_object_name=None, request_path=None,
                     schema_id=None, **kwargs):
    """Makes a given change to a MRS service

    Args:
        change_type (int): Type of change
        db_object_name (str): The name of the db_object
        request_path (str): The request_path of the db_object
        schema_id (int): The id of the schema
        **kwargs: Additional options

    Keyword Args:
        db_object_id (int): The id of the db_object
        session (object): The database session to use
        interactive (bool): Indicates whether to execute in interactive mode

    Returns:
        The result message as string
    """

    db_object_id = kwargs.get("db_object_id")
    session = kwargs.get("session")
    interactive = kwargs.get("interactive", True)

    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 db_objects to be changed
        db_object_ids = [db_object_id] if db_object_id else []
        include_enable_state = None

        schema = mrs_schemas.get_schema(
            schema_id=schema_id, session=session, interactive=interactive,
            return_formatted=False)

        if not db_object_id:
            if not db_object_name and not request_path and interactive:
                db_objects = get_db_objects(
                    schema_id=schema.get("id"),
                    include_enable_state=include_enable_state,
                    session=session, interactive=False)
                caption = ("Please select a db_object index, type "
                           "the request_path or type '*' "
                           "to select all: ")
                selection = core.prompt_for_list_item(
                    item_list=db_objects,
                    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.")

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

            if db_object_name:
                # Lookup the db_object name
                res = session.run_sql(
                    """
                    SELECT id FROM mysql_rest_service_metadata.db_object
                    WHERE name = ? AND schema_id = ?
                    """,
                    [db_object_name, schema.get("id")])
                rows = res.fetch_all()
                if rows:
                    for row in rows:
                        db_object_ids.append(row.get_field("id"))

            if request_path:
                res = session.run_sql(
                    """
                    SELECT id FROM mysql_rest_service_metadata.db_object
                    WHERE request_path = ? AND schema_id = ?
                    """,
                    [request_path, schema.get("id")])
                row = res.fetch_one()
                if row:
                    db_object_ids.append(row.get_field("id"))

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

        # Update all given services
        for db_object_id in db_object_ids:
            if change_type == DB_OBJECT_DISABLE:
                sql = """
                    UPDATE mysql_rest_service_metadata.db_object
                    SET enabled = FALSE
                    WHERE id = ?
                    """
            elif change_type == DB_OBJECT_ENABLE:
                sql = """
                    UPDATE mysql_rest_service_metadata.db_object
                    SET enabled = TRUE
                    WHERE id = ?
                    """
            elif change_type == DB_OBJECT_DELETE:
                sql = """
                    DELETE FROM mysql_rest_service_metadata.db_object
                    WHERE id = ?
                    """
            else:
                raise Exception("Operation not supported")

            res = session.run_sql(sql, [db_object_id])
            if res.get_affected_row_count() != 1:
                raise Exception(
                    f"The specified db_object with id {db_object_id} was not "
                    "found.")

        if len(db_object_ids) == 1:
            msg = "The db_object has been "
        else:
            msg = "The db_objects have been "

        if change_type == DB_OBJECT_DISABLE:
            msg += "disabled."
        if change_type == DB_OBJECT_ENABLE:
            msg += "enabled."
        elif change_type == DB_OBJECT_DELETE:
            msg += "deleted."

        return msg

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

    Args:
        change_type (int): Type of change
        db_object_name (str): The name of the db_object
        request_path (str): The request_path of the db_object
        schema_id (int): The id of the schema
        **kwargs: Additional options

    Keyword Args:
        db_object_id (int): The id of the db_object
        session (object): The database session to use
        interactive (bool): Indicates whether to execute in interactive mode

    Returns:
        The result message as string
    
Makes a given change to a MRS service
Makes a given change to a MRS service
def change_db_object(change_type=None, db_object_name=None, request_path=None,
                     schema_id=None, **kwargs):
    db_object_id = kwargs.get("db_object_id")
    session = kwargs.get("session")
    interactive = kwargs.get("interactive", True)
    try:
        session = core.get_current_session(session)
        core.ensure_rds_metadata_schema(session)
        db_object_ids = [db_object_id] if db_object_id else []
        include_enable_state = None
        schema = mrs_schemas.get_schema(
            schema_id=schema_id, session=session, interactive=interactive,
            return_formatted=False)
        if not db_object_id:
            if not db_object_name and not request_path and interactive:
                db_objects = get_db_objects(
                    schema_id=schema.get("id"),
                    include_enable_state=include_enable_state,
                    session=session, interactive=False)
                caption = ("Please select a db_object index, type "
                           "the request_path or type '*' "
                           "to select all: ")
                selection = core.prompt_for_list_item(
                    item_list=db_objects,
                    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.")
                db_object_ids = [item["id"] for item in selection]
            if db_object_name:
                res = session.run_sql(
                    """
                    SELECT id FROM mysql_rest_service_metadata.db_object
                    WHERE name = ? AND schema_id = ?
                    """,
                    [db_object_name, schema.get("id")])
                rows = res.fetch_all()
                if rows:
                    for row in rows:
                        db_object_ids.append(row.get_field("id"))
            if request_path:
                res = session.run_sql(
                    """
                    SELECT id FROM mysql_rest_service_metadata.db_object
                    WHERE request_path = ? AND schema_id = ?
                    """,
                    [request_path, schema.get("id")])
                row = res.fetch_one()
                if row:
                    db_object_ids.append(row.get_field("id"))
        if len(db_object_ids) == 0:
            raise ValueError("The specified db_object was not found.")
        for db_object_id in db_object_ids:
            if change_type == DB_OBJECT_DISABLE:
                sql = """
                    UPDATE mysql_rest_service_metadata.db_object
                    SET enabled = FALSE
                    WHERE id = ?
                    """
            elif change_type == DB_OBJECT_ENABLE:
                sql = """
                    UPDATE mysql_rest_service_metadata.db_object
                    SET enabled = TRUE
                    WHERE id = ?
                    """
            elif change_type == DB_OBJECT_DELETE:
                sql = """
                    DELETE FROM mysql_rest_service_metadata.db_object
                    WHERE id = ?
                    """
            else:
                raise Exception("Operation not supported")
            res = session.run_sql(sql, [db_object_id])
            if res.get_affected_row_count() != 1:
                raise Exception(
                    f"The specified db_object with id {db_object_id} was not "
                    "found.")
        if len(db_object_ids) == 1:
            msg = "The db_object has been "
        else:
            msg = "The db_objects have been "
        if change_type == DB_OBJECT_DISABLE:
            msg += "disabled."
        if change_type == DB_OBJECT_ENABLE:
            msg += "enabled."
        elif change_type == DB_OBJECT_DELETE:
            msg += "deleted."
        return msg
    except Exception as e:
        if interactive:
            print(f"Error: {str(e)}")
        else:
            raise
def change_db_object ( change_type = None , db_object_name = None , request_path = None , schema_id = None , ** kwargs ) : db_object_id = kwargs . get ( "db_object_id" ) session = kwargs . get ( "session" ) interactive = kwargs . get ( "interactive" , True ) try : session = core . get_current_session ( session ) core . ensure_rds_metadata_schema ( session ) db_object_ids = [ db_object_id ] if db_object_id else [ ] include_enable_state = None schema = mrs_schemas . get_schema ( schema_id = schema_id , session = session , interactive = interactive , return_formatted = False ) if not db_object_id : if not db_object_name and not request_path and interactive : db_objects = get_db_objects ( schema_id = schema . get ( "id" ) , include_enable_state = include_enable_state , session = session , interactive = False ) caption = ( "Please select a db_object index, type " "the request_path or type '*' " "to select all: " ) selection = core . prompt_for_list_item ( item_list = db_objects , 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." ) db_object_ids = [ item [ "id" ] for item in selection ] if db_object_name : res = session . run_sql ( """
                    SELECT id FROM mysql_rest_service_metadata.db_object
                    WHERE name = ? AND schema_id = ?
                    """ , [ db_object_name , schema . get ( "id" ) ] ) rows = res . fetch_all ( ) if rows : for row in rows : db_object_ids . append ( row . get_field ( "id" ) ) if request_path : res = session . run_sql ( """
                    SELECT id FROM mysql_rest_service_metadata.db_object
                    WHERE request_path = ? AND schema_id = ?
                    """ , [ request_path , schema . get ( "id" ) ] ) row = res . fetch_one ( ) if row : db_object_ids . append ( row . get_field ( "id" ) ) if len ( db_object_ids ) == 0 : raise ValueError ( "The specified db_object was not found." ) for db_object_id in db_object_ids : if change_type == DB_OBJECT_DISABLE : sql = """
                    UPDATE mysql_rest_service_metadata.db_object
                    SET enabled = FALSE
                    WHERE id = ?
                    """ elif change_type == DB_OBJECT_ENABLE : sql = """
                    UPDATE mysql_rest_service_metadata.db_object
                    SET enabled = TRUE
                    WHERE id = ?
                    """ elif change_type == DB_OBJECT_DELETE : sql = """
                    DELETE FROM mysql_rest_service_metadata.db_object
                    WHERE id = ?
                    """ else : raise Exception ( "Operation not supported" ) res = session . run_sql ( sql , [ db_object_id ] ) if res . get_affected_row_count ( ) != 1 : raise Exception ( f"The specified db_object with id {db_object_id} was not " "found." ) if len ( db_object_ids ) == 1 : msg = "The db_object has been " else : msg = "The db_objects have been " if change_type == DB_OBJECT_DISABLE : msg += "disabled." if change_type == DB_OBJECT_ENABLE : msg += "enabled." elif change_type == DB_OBJECT_DELETE : msg += "deleted." return msg except Exception as e : if interactive : print ( f"Error: {str(e)}" ) else : raise
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:
        change_type (int): Type of change
        db_object_name (str): The name of the db_object
        request_path (str): The request_path of the db_object
        schema_id (int): The id of the schema
        **kwargs: Additional options

    Keyword Args:
        db_object_id (int): The id of the db_object
        session (object): The database session to use
        interactive (bool): Indicates whether to execute in interactive mode

    Returns:
        The result message as string
    """ # Make sure the MRS metadata schema exists and has the right version # The list of db_objects to be changed # Lookup the db_object name # Update all given services
The result message as string
The result message as string
change_type
Type of change
Type of change
db_object_name
The name of the db_object
The name of the db_object
request_path
The request_path of the db_object
The request_path of the db_object
schema_id
The id of the schema
The id of the schema
**kwargs
None