mrs_plugin/schemas.py
def add_schema(**kwargs):
    """Add a schema to the given MRS service

    Args:
        **kwargs: Additional options

    Keyword Args:
        schema_name (str): The name of the schema to add
        service_id (int): The id of the service the schema should be added to
        request_path (str): The request_path
        requires_auth (bool): Whether authentication is required to access
            the schema
        enabled (bool): The enabled state
        items_per_page (int): The number of items returned per page
        comments (str): Comments for the schema
        session (object): The database session to use.
        interactive (bool): Indicates whether to execute in interactive mode

    Returns:
        None
    """

    schema_name = kwargs.get("schema_name")
    service_id = kwargs.get("service_id")
    request_path = kwargs.get("request_path")
    requires_auth = kwargs.get("requires_auth")
    enabled = kwargs.get("enabled", True)
    items_per_page = kwargs.get("items_per_page")
    comments = kwargs.get("comments")
    session = kwargs.get("session")
    interactive = kwargs.get("interactive", True)

    try:
        session = core.get_current_session(session)
        service = core.get_current_service()
        if service_id is not None or service is None:
            service = mrs_services.get_service(service_id=service_id, get_default=False if service_id else True,
                auto_select_single=True,
                session=session, interactive=interactive,
                return_formatted=False)

        if not schema_name and interactive:
            res = session.run_sql('''
                SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA
                WHERE SCHEMA_NAME NOT LIKE ?
                    AND SCHEMA_NAME NOT LIKE ?
                    AND SCHEMA_NAME <> ?
                ''', ['.%', 'mysql_%', 'mysql'])
            rows = res.fetch_all()
            schemas = \
                [row.get_field("SCHEMA_NAME") for row in rows]

            if len(schemas) == 0:
                raise ValueError('No database schemas available.')

            schema_name = core.prompt_for_list_item(
                item_list=schemas,
                prompt_caption='Please enter the name or index of a schema: ',
                print_list=True)

            if not schema_name:
                raise ValueError('Operation cancelled.')
        # If a schema name has been provided, check if that schema exists
        else:
            res = session.run_sql('''
                SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA
                WHERE SCHEMA_NAME = ?
                ORDER BY SCHEMA_NAME
                ''', [schema_name])
            row = res.fetch_one()
            if row:
                schema_name = row.get_field("SCHEMA_NAME")
            else:
                raise ValueError(
                    f"The given schema_name '{schema_name}' "
                    "does not exists.")

        # Get request_path
        if not request_path:
            if interactive:
                request_path = prompt_for_request_path(schema_name)
            else:
                request_path = '/' + schema_name

        if not request_path.startswith('/'):
            raise Exception("The request_path has to start with '/'.")

        core.check_request_path(
            service.get('url_host_name') + service.get('url_context_root') +
            request_path,
            session=session)

        # Get requires_auth
        if not requires_auth:
            if interactive:
                requires_auth = prompt_for_requires_auth()
            else:
                requires_auth = False

        # Get items_per_page
        if not items_per_page:
            if interactive:
                items_per_page = prompt_for_items_per_page()
            else:
                items_per_page = 25

        # Get comments
        if not comments:
            if interactive:
                comments = core.prompt(
                    "Comments: ").strip()
            else:
                comments = ""

        res = session.run_sql("""
            INSERT INTO `mysql_rest_service_metadata`.db_schema(
                service_id, name, request_path,
                requires_auth, enabled, items_per_page, comments)
            VALUES(?, ?, ?, ?, ?, ?, ?)
            """, [service.get("id"), schema_name,
                  request_path, 1 if requires_auth else 0,
                  1 if enabled else 0,
                  items_per_page, comments])

        if interactive:
            return "\n" + "Schema added successfully."
        else:
            return res.auto_increment_value

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

    Args:
        **kwargs: Additional options

    Keyword Args:
        schema_name (str): The name of the schema to add
        service_id (int): The id of the service the schema should be added to
        request_path (str): The request_path
        requires_auth (bool): Whether authentication is required to access
            the schema
        enabled (bool): The enabled state
        items_per_page (int): The number of items returned per page
        comments (str): Comments for the schema
        session (object): The database session to use.
        interactive (bool): Indicates whether to execute in interactive mode

    Returns:
        None
    
Add a schema to the given MRS service
Add a schema to the given MRS service
def add_schema(**kwargs):
    schema_name = kwargs.get("schema_name")
    service_id = kwargs.get("service_id")
    request_path = kwargs.get("request_path")
    requires_auth = kwargs.get("requires_auth")
    enabled = kwargs.get("enabled", True)
    items_per_page = kwargs.get("items_per_page")
    comments = kwargs.get("comments")
    session = kwargs.get("session")
    interactive = kwargs.get("interactive", True)
    try:
        session = core.get_current_session(session)
        service = core.get_current_service()
        if service_id is not None or service is None:
            service = mrs_services.get_service(service_id=service_id, get_default=False if service_id else True,
                auto_select_single=True,
                session=session, interactive=interactive,
                return_formatted=False)
        if not schema_name and interactive:
            res = session.run_sql('''
                SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA
                WHERE SCHEMA_NAME NOT LIKE ?
                    AND SCHEMA_NAME NOT LIKE ?
                    AND SCHEMA_NAME <> ?
                ''', ['.%', 'mysql_%', 'mysql'])
            rows = res.fetch_all()
            schemas = \
                [row.get_field("SCHEMA_NAME") for row in rows]
            if len(schemas) == 0:
                raise ValueError('No database schemas available.')
            schema_name = core.prompt_for_list_item(
                item_list=schemas,
                prompt_caption='Please enter the name or index of a schema: ',
                print_list=True)
            if not schema_name:
                raise ValueError('Operation cancelled.')
        else:
            res = session.run_sql('''
                SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA
                WHERE SCHEMA_NAME = ?
                ORDER BY SCHEMA_NAME
                ''', [schema_name])
            row = res.fetch_one()
            if row:
                schema_name = row.get_field("SCHEMA_NAME")
            else:
                raise ValueError(
                    f"The given schema_name '{schema_name}' "
                    "does not exists.")
        if not request_path:
            if interactive:
                request_path = prompt_for_request_path(schema_name)
            else:
                request_path = '/' + schema_name
        if not request_path.startswith('/'):
            raise Exception("The request_path has to start with '/'.")
        core.check_request_path(
            service.get('url_host_name') + service.get('url_context_root') +
            request_path,
            session=session)
        if not requires_auth:
            if interactive:
                requires_auth = prompt_for_requires_auth()
            else:
                requires_auth = False
        if not items_per_page:
            if interactive:
                items_per_page = prompt_for_items_per_page()
            else:
                items_per_page = 25
        if not comments:
            if interactive:
                comments = core.prompt(
                    "Comments: ").strip()
            else:
                comments = ""
        res = session.run_sql("""
            INSERT INTO `mysql_rest_service_metadata`.db_schema(
                service_id, name, request_path,
                requires_auth, enabled, items_per_page, comments)
            VALUES(?, ?, ?, ?, ?, ?, ?)
            """, [service.get("id"), schema_name,
                  request_path, 1 if requires_auth else 0,
                  1 if enabled else 0,
                  items_per_page, comments])
        if interactive:
            return "\n" + "Schema added successfully."
        else:
            return res.auto_increment_value
    except Exception as e:
        if interactive:
            print(f"Error: {str(e)}")
        else:
            raise
def add_schema ( ** kwargs ) : schema_name = kwargs . get ( "schema_name" ) service_id = kwargs . get ( "service_id" ) request_path = kwargs . get ( "request_path" ) requires_auth = kwargs . get ( "requires_auth" ) enabled = kwargs . get ( "enabled" , True ) items_per_page = kwargs . get ( "items_per_page" ) comments = kwargs . get ( "comments" ) session = kwargs . get ( "session" ) interactive = kwargs . get ( "interactive" , True ) try : session = core . get_current_session ( session ) service = core . get_current_service ( ) if service_id is not None or service is None : service = mrs_services . get_service ( service_id = service_id , get_default = False if service_id else True , auto_select_single = True , session = session , interactive = interactive , return_formatted = False ) if not schema_name and interactive : res = session . run_sql ( '''
                SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA
                WHERE SCHEMA_NAME NOT LIKE ?
                    AND SCHEMA_NAME NOT LIKE ?
                    AND SCHEMA_NAME <> ?
                ''' , [ '.%' , 'mysql_%' , 'mysql' ] ) rows = res . fetch_all ( ) schemas = [ row . get_field ( "SCHEMA_NAME" ) for row in rows ] if len ( schemas ) == 0 : raise ValueError ( 'No database schemas available.' ) schema_name = core . prompt_for_list_item ( item_list = schemas , prompt_caption = 'Please enter the name or index of a schema: ' , print_list = True ) if not schema_name : raise ValueError ( 'Operation cancelled.' ) else : res = session . run_sql ( '''
                SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA
                WHERE SCHEMA_NAME = ?
                ORDER BY SCHEMA_NAME
                ''' , [ schema_name ] ) row = res . fetch_one ( ) if row : schema_name = row . get_field ( "SCHEMA_NAME" ) else : raise ValueError ( f"The given schema_name '{schema_name}' " "does not exists." ) if not request_path : if interactive : request_path = prompt_for_request_path ( schema_name ) else : request_path = '/' + schema_name if not request_path . startswith ( '/' ) : raise Exception ( "The request_path has to start with '/'." ) core . check_request_path ( service . get ( 'url_host_name' ) + service . get ( 'url_context_root' ) + request_path , session = session ) if not requires_auth : if interactive : requires_auth = prompt_for_requires_auth ( ) else : requires_auth = False if not items_per_page : if interactive : items_per_page = prompt_for_items_per_page ( ) else : items_per_page = 25 if not comments : if interactive : comments = core . prompt ( "Comments: " ) . strip ( ) else : comments = "" res = session . run_sql ( """
            INSERT INTO `mysql_rest_service_metadata`.db_schema(
                service_id, name, request_path,
                requires_auth, enabled, items_per_page, comments)
            VALUES(?, ?, ?, ?, ?, ?, ?)
            """ , [ service . get ( "id" ) , schema_name , request_path , 1 if requires_auth else 0 , 1 if enabled else 0 , items_per_page , comments ] ) if interactive : return "\n" + "Schema added successfully." else : return res . auto_increment_value except Exception as e : if interactive : print ( f"Error: {str(e)}" ) else : raise
Add a schema to the given MRS service
Add a schema to the given MRS service
"""Add a schema to the given MRS service

    Args:
        **kwargs: Additional options

    Keyword Args:
        schema_name (str): The name of the schema to add
        service_id (int): The id of the service the schema should be added to
        request_path (str): The request_path
        requires_auth (bool): Whether authentication is required to access
            the schema
        enabled (bool): The enabled state
        items_per_page (int): The number of items returned per page
        comments (str): Comments for the schema
        session (object): The database session to use.
        interactive (bool): Indicates whether to execute in interactive mode

    Returns:
        None
    """ # If a schema name has been provided, check if that schema exists # Get request_path # Get requires_auth # Get items_per_page # Get comments
None
**kwargs
None