mrs_plugin/content_sets.py
def change_content_set(change_type=None, request_path=None, service_id=None,
                       **kwargs):
    """Makes a given change to a MRS content set

    Args:
        change_type (int): Type of change
        request_path (str): The request_path of the content_set
        service_id (int): The id of the service
        **kwargs: Additional options

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

    Returns:
        The result message as string
    """

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

    try:
        session = core.get_current_session(session)

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

        # The list of content_sets to be changed
        content_set_ids = [content_set_id] if content_set_id else []
        include_enable_state = None

        if not content_set_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 request_path and interactive:
                content_sets = get_content_sets(
                    service_id=service.get("id"),
                    include_enable_state=include_enable_state,
                    session=session, interactive=False)
                caption = ("Please select an index, type "
                           "the request_path or type '*' "
                           "to select all: ")
                selection = core.prompt_for_list_item(
                    item_list=content_sets,
                    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.")

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

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

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

        # Update all given services
        for content_set_id in content_set_ids:
            if change_type == CONTENT_SET_DISABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.`content_set`
                    SET enabled = FALSE
                    WHERE id = ?
                    """
            elif change_type == CONTENT_SET_ENABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.`content_set`
                    SET enabled = TRUE
                    WHERE id = ?
                    """
            elif change_type == CONTENT_SET_DELETE:
                res = session.run_sql("""
                    DELETE FROM `mysql_rest_service_metadata`.`content_file`
                    WHERE content_set_id = ?
                    """, [content_set_id])
                sql = """
                    DELETE FROM `mysql_rest_service_metadata`.`content_set`
                    WHERE id = ?
                    """
            else:
                raise Exception("Operation not supported")

            res = session.run_sql(sql, [content_set_id])
            if res.get_affected_row_count() == 0:
                raise Exception(
                    f"The specified content_set with id {content_set_id} was "
                    "not found.")

        if len(content_set_ids) == 1:
            msg = "The content set has been "
        else:
            msg = "The content sets have been "

        if change_type == CONTENT_SET_DISABLE:
            msg += "disabled."
        if change_type == CONTENT_SET_ENABLE:
            msg += "enabled."
        elif change_type == CONTENT_SET_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 content set

    Args:
        change_type (int): Type of change
        request_path (str): The request_path of the content_set
        service_id (int): The id of the service
        **kwargs: Additional options

    Keyword Args:
        content_set_id (int): The id of the content_set
        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 content set
Makes a given change to a MRS content set
def change_content_set(change_type=None, request_path=None, service_id=None,
                       **kwargs):
    content_set_id = kwargs.get("content_set_id")
    session = kwargs.get("session")
    interactive = kwargs.get("interactive", True)
    try:
        session = core.get_current_session(session)
        core.ensure_rds_metadata_schema(session)
        content_set_ids = [content_set_id] if content_set_id else []
        include_enable_state = None
        if not content_set_id:
            service = mrs_services.get_service(
                service_id=service_id, session=session, interactive=interactive,
                return_formatted=False)
            if not request_path and interactive:
                content_sets = get_content_sets(
                    service_id=service.get("id"),
                    include_enable_state=include_enable_state,
                    session=session, interactive=False)
                caption = ("Please select an index, type "
                           "the request_path or type '*' "
                           "to select all: ")
                selection = core.prompt_for_list_item(
                    item_list=content_sets,
                    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.")
                content_set_ids = [item["id"] for item in selection]
            if request_path:
                res = session.run_sql(
                    """
                    SELECT id FROM `mysql_rest_service_metadata`.`content_set`
                    WHERE request_path = ? AND service_id = ?
                    """,
                    [request_path, service.get("id")])
                row = res.fetch_one()
                if row:
                    content_set_ids.append(row.get_field("id"))
        if len(content_set_ids) == 0:
            raise ValueError("The specified content_set was not found.")
        for content_set_id in content_set_ids:
            if change_type == CONTENT_SET_DISABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.`content_set`
                    SET enabled = FALSE
                    WHERE id = ?
                    """
            elif change_type == CONTENT_SET_ENABLE:
                sql = """
                    UPDATE `mysql_rest_service_metadata`.`content_set`
                    SET enabled = TRUE
                    WHERE id = ?
                    """
            elif change_type == CONTENT_SET_DELETE:
                res = session.run_sql("""
                    DELETE FROM `mysql_rest_service_metadata`.`content_file`
                    WHERE content_set_id = ?
                    """, [content_set_id])
                sql = """
                    DELETE FROM `mysql_rest_service_metadata`.`content_set`
                    WHERE id = ?
                    """
            else:
                raise Exception("Operation not supported")
            res = session.run_sql(sql, [content_set_id])
            if res.get_affected_row_count() == 0:
                raise Exception(
                    f"The specified content_set with id {content_set_id} was "
                    "not found.")
        if len(content_set_ids) == 1:
            msg = "The content set has been "
        else:
            msg = "The content sets have been "
        if change_type == CONTENT_SET_DISABLE:
            msg += "disabled."
        if change_type == CONTENT_SET_ENABLE:
            msg += "enabled."
        elif change_type == CONTENT_SET_DELETE:
            msg += "deleted."
        return msg
    except Exception as e:
        if interactive:
            print(f"Error: {str(e)}")
        else:
            raise
def change_content_set ( change_type = None , request_path = None , service_id = None , ** kwargs ) : content_set_id = kwargs . get ( "content_set_id" ) session = kwargs . get ( "session" ) interactive = kwargs . get ( "interactive" , True ) try : session = core . get_current_session ( session ) core . ensure_rds_metadata_schema ( session ) content_set_ids = [ content_set_id ] if content_set_id else [ ] include_enable_state = None if not content_set_id : service = mrs_services . get_service ( service_id = service_id , session = session , interactive = interactive , return_formatted = False ) if not request_path and interactive : content_sets = get_content_sets ( service_id = service . get ( "id" ) , include_enable_state = include_enable_state , session = session , interactive = False ) caption = ( "Please select an index, type " "the request_path or type '*' " "to select all: " ) selection = core . prompt_for_list_item ( item_list = content_sets , 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." ) content_set_ids = [ item [ "id" ] for item in selection ] if request_path : res = session . run_sql ( """
                    SELECT id FROM `mysql_rest_service_metadata`.`content_set`
                    WHERE request_path = ? AND service_id = ?
                    """ , [ request_path , service . get ( "id" ) ] ) row = res . fetch_one ( ) if row : content_set_ids . append ( row . get_field ( "id" ) ) if len ( content_set_ids ) == 0 : raise ValueError ( "The specified content_set was not found." ) for content_set_id in content_set_ids : if change_type == CONTENT_SET_DISABLE : sql = """
                    UPDATE `mysql_rest_service_metadata`.`content_set`
                    SET enabled = FALSE
                    WHERE id = ?
                    """ elif change_type == CONTENT_SET_ENABLE : sql = """
                    UPDATE `mysql_rest_service_metadata`.`content_set`
                    SET enabled = TRUE
                    WHERE id = ?
                    """ elif change_type == CONTENT_SET_DELETE : res = session . run_sql ( """
                    DELETE FROM `mysql_rest_service_metadata`.`content_file`
                    WHERE content_set_id = ?
                    """ , [ content_set_id ] ) sql = """
                    DELETE FROM `mysql_rest_service_metadata`.`content_set`
                    WHERE id = ?
                    """ else : raise Exception ( "Operation not supported" ) res = session . run_sql ( sql , [ content_set_id ] ) if res . get_affected_row_count ( ) == 0 : raise Exception ( f"The specified content_set with id {content_set_id} was " "not found." ) if len ( content_set_ids ) == 1 : msg = "The content set has been " else : msg = "The content sets have been " if change_type == CONTENT_SET_DISABLE : msg += "disabled." if change_type == CONTENT_SET_ENABLE : msg += "enabled." elif change_type == CONTENT_SET_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 content set
Makes a given change to a MRS content set
"""Makes a given change to a MRS content set

    Args:
        change_type (int): Type of change
        request_path (str): The request_path of the content_set
        service_id (int): The id of the service
        **kwargs: Additional options

    Keyword Args:
        content_set_id (int): The id of the content_set
        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 content_set exists and has the right version # The list of content_sets to be changed # Check if given service_id exists or use the current service # Lookup the content_set name # Update all given services
The result message as string
The result message as string
change_type
Type of change
Type of change
request_path
The request_path of the content_set
The request_path of the content_set
service_id
The id of the service
The id of the service
**kwargs
None