(ctx, **kwargs)
\b
    - DESCRIPTION:
        Deploy model.

    \b
    - USAGE:
        flow model deploy --model-id $MODEL_ID --model-version $MODEL_VERSION
\b
    - DESCRIPTION:
        Deploy model.
\ b - DESCRIPTION : Deploy model .
def deploy(ctx, **kwargs):
    """
    \b
    - DESCRIPTION:
        Deploy model.

    \b
    - USAGE:
        flow model deploy --model-id $MODEL_ID --model-version $MODEL_VERSION

    """
    request_data = {
        'model_id': kwargs['model_id'],
        'model_version': kwargs['model_version'],
    }

    if kwargs.get("cpn_list") or kwargs.get("cpn_path"):
        if kwargs.get("cpn_list"):
            cpn_str = kwargs["cpn_list"]
        elif kwargs.get("cpn_path"):
            with open(kwargs["cpn_path"], "r") as fp:
                cpn_str = fp.read()
        else:
            cpn_str = ""

        if isinstance(cpn_str, list):
            cpn_list = cpn_str
        else:
            if (cpn_str.find("/") and cpn_str.find("\\")) != -1:
                raise Exception("Component list string should not contain '/' or '\\'.")
            cpn_str = cpn_str.replace(" ", "").replace("\n", "").strip(",[]")
            cpn_list = cpn_str.split(",")
        request_data['cpn_list'] = cpn_list
    elif kwargs.get("dsl_path"):
        with open(kwargs["dsl_path"], "r") as ft:
            predict_dsl = ft.read()
        request_data['dsl'] = predict_dsl

    request_data['components_checkpoint'] = {}
    for i in ('cpn_step_index', 'cpn_step_name'):
        for j in kwargs[i]:
            component, checkpoint = j.rsplit(':', 1)

            if i == 'cpn_step_index':
                checkpoint = int(checkpoint)
            if component in request_data['components_checkpoint']:
                raise KeyError(f"Duplicated component name '{component}'.")

            request_data['components_checkpoint'][component] = {
                i[4:]: checkpoint,
            }

    config_data, dsl_data = preprocess(**request_data)
    access_server('post', ctx, 'model/deploy', config_data)
def deploy ( ctx , * * kwargs ) : request_data = { 'model_id' : kwargs [ 'model_id' ] , 'model_version' : kwargs [ 'model_version' ] , } if kwargs . get ( "cpn_list" ) or kwargs . get ( "cpn_path" ) : if kwargs . get ( "cpn_list" ) : cpn_str = kwargs [ "cpn_list" ] elif kwargs . get ( "cpn_path" ) : with open ( kwargs [ "cpn_path" ] , "r" ) as fp : cpn_str = fp . read ( ) else : cpn_str = "" if isinstance ( cpn_str , list ) : cpn_list = cpn_str else : if ( cpn_str . find ( "/" ) and cpn_str . find ( "\\" ) ) != - 1 : raise Exception ( "Component list string should not contain '/' or '\\'." ) cpn_str = cpn_str . replace ( " " , "" ) . replace ( "\n" , "" ) . strip ( ",[]" ) cpn_list = cpn_str . split ( "," ) request_data [ 'cpn_list' ] = cpn_list elif kwargs . get ( "dsl_path" ) : with open ( kwargs [ "dsl_path" ] , "r" ) as ft : predict_dsl = ft . read ( ) request_data [ 'dsl' ] = predict_dsl request_data [ 'components_checkpoint' ] = { } for i in ( 'cpn_step_index' , 'cpn_step_name' ) : for j in kwargs [ i ] : component , checkpoint = j . rsplit ( ':' , 1 ) if i == 'cpn_step_index' : checkpoint = int ( checkpoint ) if component in request_data [ 'components_checkpoint' ] : raise KeyError ( f"Duplicated component name '{component}'." ) request_data [ 'components_checkpoint' ] [ component ] = { i [ 4 : ] : checkpoint , } config_data , dsl_data = preprocess ( * * request_data ) access_server ( 'post' , ctx , 'model/deploy' , config_data )