mrs_plugin/general.py
def cd(path=None, session=None):
    """Change the current service and/or schema

    Args:
        path (str): The name of the service and/or schema to make current
        session (object): The database session to use

    Returns:
        none

    """

    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)

        # If a path was given, try to get the corresponding objects
        if path:
            service, schema, content_set = core.analyze_service_path(
                path=path, session=session)

            current_path = ""
            if service:
                current_path += service.get("host_ctx")
            if schema:
                current_path += schema.get("request_path")
            if content_set:
                current_path += content_set.get("request_path")

            core.set_current_objects(service=service, schema=schema,
                                content_set=content_set)

            print(f"Current path set to {current_path}")
            return
        # If an empty path was given, set the current path to root
        elif path == "":
            core.set_current_objects(service=None, schema=None, db_object=None)
            print(f"Current path set to root.")
        # If no path was given, let the user choose
        else:
            current_service = core.get_current_service(session=session)

            # If the current service has not been set, let the user select it
            if not current_service:
                # Check if there already is at least one service
                res = session.run_sql("""
                    SELECT COUNT(*) as service_count
                    FROM `mysql_rest_service_metadata`.service
                    """)
                row = res.fetch_one()
                service_count = row.get_field("service_count") if row else 0

                # If there is exactly one service, set id to its id
                if service_count == 0:
                    raise ValueError("No MRS service available.")

                # If there are more services, let the user select one or all
                print("MRS - Service Listing\n")
                services = mrs_services.get_services(
                    session=session, interactive=False)
                item = core.prompt_for_list_item(
                    item_list=services,
                    prompt_caption=("Please select a service index or type "
                                    "'hostname/root_context'"),
                    item_name_property="host_ctx",
                    given_value=None,
                    print_list=True)
                if not item:
                    raise ValueError("Operation cancelled.")
                else:
                    core.set_current_objects(service=item)
                    print(f"Current path set to {item.get('host_ctx')}")
                    return

            current_schema = core.get_current_schema(session=session)
            current_content_set = core.get_current_content_set(session=session)

            if not current_schema and not current_content_set:
                print(f"MRS Service {current_service.get('host_ctx')} - "
                      "Schema and Content Set Listing:\n")
                # Get the schema
                schemas = mrs_schemas.get_schemas(
                    service_id=current_service.get("id"),
                    session=session, interactive=False, return_formatted=False)
                context_sets = mrs_content_sets.get_content_sets(
                    service_id=current_service.get("id"),
                    session=session, interactive=False)

                items = schemas + context_sets

                selection = core.prompt_for_list_item(
                    item_list=items, prompt_caption=("Please select a schema "
                                                     "or content set: "),
                    item_name_property='request_path', print_list=True)

                # If there is a key called name, it is a schema
                if 'name' in selection:
                    core.set_current_objects(service=current_service,
                                        schema=selection)
                else:
                    core.set_current_objects(service=current_service,
                                        content_set=selection)

                if selection:
                    print(f"Current path set to {selection.get('host_ctx')}"
                          f"{selection.get('request_path')}")
                    return

    except Exception as e:
        print(f"Error: {str(e)}")
Change the current service and/or schema

    Args:
        path (str): The name of the service and/or schema to make current
        session (object): The database session to use

    Returns:
        none

    
Change the current service and/or schema
Change the current service and / or schema
def cd(path=None, session=None):
    try:
        session = core.get_current_session(session)
        core.ensure_rds_metadata_schema(session)
        if path:
            service, schema, content_set = core.analyze_service_path(
                path=path, session=session)
            current_path = ""
            if service:
                current_path += service.get("host_ctx")
            if schema:
                current_path += schema.get("request_path")
            if content_set:
                current_path += content_set.get("request_path")
            core.set_current_objects(service=service, schema=schema,
                                content_set=content_set)
            print(f"Current path set to {current_path}")
            return
        elif path == "":
            core.set_current_objects(service=None, schema=None, db_object=None)
            print(f"Current path set to root.")
        else:
            current_service = core.get_current_service(session=session)
            if not current_service:
                res = session.run_sql("""
                    SELECT COUNT(*) as service_count
                    FROM `mysql_rest_service_metadata`.service
                    """)
                row = res.fetch_one()
                service_count = row.get_field("service_count") if row else 0
                if service_count == 0:
                    raise ValueError("No MRS service available.")
                print("MRS - Service Listing\n")
                services = mrs_services.get_services(
                    session=session, interactive=False)
                item = core.prompt_for_list_item(
                    item_list=services,
                    prompt_caption=("Please select a service index or type "
                                    "'hostname/root_context'"),
                    item_name_property="host_ctx",
                    given_value=None,
                    print_list=True)
                if not item:
                    raise ValueError("Operation cancelled.")
                else:
                    core.set_current_objects(service=item)
                    print(f"Current path set to {item.get('host_ctx')}")
                    return
            current_schema = core.get_current_schema(session=session)
            current_content_set = core.get_current_content_set(session=session)
            if not current_schema and not current_content_set:
                print(f"MRS Service {current_service.get('host_ctx')} - "
                      "Schema and Content Set Listing:\n")
                schemas = mrs_schemas.get_schemas(
                    service_id=current_service.get("id"),
                    session=session, interactive=False, return_formatted=False)
                context_sets = mrs_content_sets.get_content_sets(
                    service_id=current_service.get("id"),
                    session=session, interactive=False)
                items = schemas + context_sets
                selection = core.prompt_for_list_item(
                    item_list=items, prompt_caption=("Please select a schema "
                                                     "or content set: "),
                    item_name_property='request_path', print_list=True)
                if 'name' in selection:
                    core.set_current_objects(service=current_service,
                                        schema=selection)
                else:
                    core.set_current_objects(service=current_service,
                                        content_set=selection)
                if selection:
                    print(f"Current path set to {selection.get('host_ctx')}"
                          f"{selection.get('request_path')}")
                    return
    except Exception as e:
        print(f"Error: {str(e)}")
def cd ( path = None , session = None ) : try : session = core . get_current_session ( session ) core . ensure_rds_metadata_schema ( session ) if path : service , schema , content_set = core . analyze_service_path ( path = path , session = session ) current_path = "" if service : current_path += service . get ( "host_ctx" ) if schema : current_path += schema . get ( "request_path" ) if content_set : current_path += content_set . get ( "request_path" ) core . set_current_objects ( service = service , schema = schema , content_set = content_set ) print ( f"Current path set to {current_path}" ) return elif path == "" : core . set_current_objects ( service = None , schema = None , db_object = None ) print ( f"Current path set to root." ) else : current_service = core . get_current_service ( session = session ) if not current_service : res = session . run_sql ( """
                    SELECT COUNT(*) as service_count
                    FROM `mysql_rest_service_metadata`.service
                    """ ) row = res . fetch_one ( ) service_count = row . get_field ( "service_count" ) if row else 0 if service_count == 0 : raise ValueError ( "No MRS service available." ) print ( "MRS - Service Listing\n" ) services = mrs_services . get_services ( session = session , interactive = False ) item = core . prompt_for_list_item ( item_list = services , prompt_caption = ( "Please select a service index or type " "'hostname/root_context'" ) , item_name_property = "host_ctx" , given_value = None , print_list = True ) if not item : raise ValueError ( "Operation cancelled." ) else : core . set_current_objects ( service = item ) print ( f"Current path set to {item.get('host_ctx')}" ) return current_schema = core . get_current_schema ( session = session ) current_content_set = core . get_current_content_set ( session = session ) if not current_schema and not current_content_set : print ( f"MRS Service {current_service.get('host_ctx')} - " "Schema and Content Set Listing:\n" ) schemas = mrs_schemas . get_schemas ( service_id = current_service . get ( "id" ) , session = session , interactive = False , return_formatted = False ) context_sets = mrs_content_sets . get_content_sets ( service_id = current_service . get ( "id" ) , session = session , interactive = False ) items = schemas + context_sets selection = core . prompt_for_list_item ( item_list = items , prompt_caption = ( "Please select a schema " "or content set: " ) , item_name_property = 'request_path' , print_list = True ) if 'name' in selection : core . set_current_objects ( service = current_service , schema = selection ) else : core . set_current_objects ( service = current_service , content_set = selection ) if selection : print ( f"Current path set to {selection.get('host_ctx')}" f"{selection.get('request_path')}" ) return except Exception as e : print ( f"Error: {str(e)}" )
Change the current service and/or schema
Change the current service and / or schema
"""Change the current service and/or schema

    Args:
        path (str): The name of the service and/or schema to make current
        session (object): The database session to use

    Returns:
        none

    """ # Make sure the MRS metadata schema exists and has the right version
 # If a path was given, try to get the corresponding objects
 # If an empty path was given, set the current path to root
 # If no path was given, let the user choose
 # If the current service has not been set, let the user select it
 # Check if there already is at least one service
 # If there is exactly one service, set id to its id
 # If there are more services, let the user select one or all
 # Get the schema
 # If there is a key called name, it is a schema

None
path
The name of the service and/or schema to make current
The name of the service and / or schema to make current
session
The database session to use
The database session to use