def create_synccopy_specifications(ctx_cli_options, config):
    # type: (dict, dict) -> List[blobxfer.models.synccopy.Specification]
    """Create a list of SyncCopy Specification objects from configuration
    :param dict ctx_cli_options: cli options
    :param dict config: config dict
    :rtype: list
    :return: list of SyncCopy Specification objects
    """
    cli_conf = ctx_cli_options[ctx_cli_options['_action']]
    cli_options = cli_conf['options']
    specs = []
    for conf in config['synccopy']:
        if 'options' in conf:
            conf_options = conf['options']
        else:
            conf_options = {}
        # get source mode
        mode = _merge_setting(
            cli_options, conf_options, 'mode', default='auto').lower()
        if mode == 'auto':
            mode = blobxfer.models.azure.StorageModes.Auto
        elif mode == 'append':
            mode = blobxfer.models.azure.StorageModes.Append
        elif mode == 'block':
            mode = blobxfer.models.azure.StorageModes.Block
        elif mode == 'file':
            mode = blobxfer.models.azure.StorageModes.File
        elif mode == 'page':
            mode = blobxfer.models.azure.StorageModes.Page
        else:
            raise ValueError('unknown source mode: {}'.format(mode))
        # get destination mode
        destmode = _merge_setting(
            cli_options, conf_options, 'dest_mode', name_cli='dest_mode')
        if blobxfer.util.is_none_or_empty(destmode):
            destmode = mode
        else:
            destmode = destmode.lower()
            if destmode == 'auto':
                destmode = blobxfer.models.azure.StorageModes.Auto
            elif destmode == 'append':
                destmode = blobxfer.models.azure.StorageModes.Append
            elif destmode == 'block':
                destmode = blobxfer.models.azure.StorageModes.Block
            elif destmode == 'file':
                destmode = blobxfer.models.azure.StorageModes.File
            elif destmode == 'page':
                destmode = blobxfer.models.azure.StorageModes.Page
            else:
                raise ValueError('unknown dest mode: {}'.format(destmode))
        # create specification
        conf_sod = conf_options.get('skip_on', {})
        cli_sod = cli_options['skip_on']
        scs = blobxfer.models.synccopy.Specification(
            synccopy_options=blobxfer.models.options.SyncCopy(
                access_tier=_merge_setting(
                    cli_options, conf_options, 'access_tier', default=None),
                delete_extraneous_destination=_merge_setting(
                    cli_options, conf_options,
                    'delete_extraneous_destination', default=False),
                dest_mode=destmode,
                mode=mode,
                overwrite=_merge_setting(
                    cli_options, conf_options, 'overwrite', default=True),
                recursive=_merge_setting(
                    cli_options, conf_options, 'recursive', default=True),
                rename=_merge_setting(
                    cli_options, conf_options, 'rename', default=False),
            ),
            skip_on_options=blobxfer.models.options.SkipOn(
                filesize_match=_merge_setting(
                    cli_sod, conf_sod, 'filesize_match', default=False),
                lmt_ge=_merge_setting(
                    cli_sod, conf_sod, 'lmt_ge', default=False),
                md5_match=_merge_setting(
                    cli_sod, conf_sod, 'md5_match', default=False),
            ),
        )
        # create remote source paths
        for src in conf['source']:
            sa = next(iter(src))
            asp = blobxfer.operations.azure.SourcePath()
            asp.add_path_with_storage_account(src[sa], sa)
            incl = _merge_setting(cli_conf, conf, 'include', default=None)
            if blobxfer.util.is_not_empty(incl):
                asp.add_includes(incl)
            excl = _merge_setting(cli_conf, conf, 'exclude', default=None)
            if blobxfer.util.is_not_empty(excl):
                asp.add_excludes(excl)
            scs.add_azure_source_path(asp)
        # create remote destination paths
        for dst in conf['destination']:
            if len(dst) != 1:
                raise RuntimeError(
                    'invalid number of destination pairs specified per entry')
            sa = next(iter(dst))
            adp = blobxfer.operations.azure.DestinationPath()
            adp.add_path_with_storage_account(dst[sa], sa)
            scs.add_azure_destination_path(adp)
        # append spec to list
        specs.append(scs)
    return specs
def create_synccopy_specifications ( ctx_cli_options , config ) : cli_conf = ctx_cli_options [ ctx_cli_options [ '_action' ] ] cli_options = cli_conf [ 'options' ] specs = [ ] for conf in config [ 'synccopy' ] : if 'options' in conf : conf_options = conf [ 'options' ] else : conf_options = { } mode = _merge_setting ( cli_options , conf_options , 'mode' , default = 'auto' ) . lower ( ) if mode == 'auto' : mode = blobxfer . models . azure . StorageModes . Auto elif mode == 'append' : mode = blobxfer . models . azure . StorageModes . Append elif mode == 'block' : mode = blobxfer . models . azure . StorageModes . Block elif mode == 'file' : mode = blobxfer . models . azure . StorageModes . File elif mode == 'page' : mode = blobxfer . models . azure . StorageModes . Page else : raise ValueError ( 'unknown source mode: {}' . format ( mode ) ) destmode = _merge_setting ( cli_options , conf_options , 'dest_mode' , name_cli = 'dest_mode' ) if blobxfer . util . is_none_or_empty ( destmode ) : destmode = mode else : destmode = destmode . lower ( ) if destmode == 'auto' : destmode = blobxfer . models . azure . StorageModes . Auto elif destmode == 'append' : destmode = blobxfer . models . azure . StorageModes . Append elif destmode == 'block' : destmode = blobxfer . models . azure . StorageModes . Block elif destmode == 'file' : destmode = blobxfer . models . azure . StorageModes . File elif destmode == 'page' : destmode = blobxfer . models . azure . StorageModes . Page else : raise ValueError ( 'unknown dest mode: {}' . format ( destmode ) ) conf_sod = conf_options . get ( 'skip_on' , { } ) cli_sod = cli_options [ 'skip_on' ] scs = blobxfer . models . synccopy . Specification ( synccopy_options = blobxfer . models . options . SyncCopy ( access_tier = _merge_setting ( cli_options , conf_options , 'access_tier' , default = None ) , delete_extraneous_destination = _merge_setting ( cli_options , conf_options , 'delete_extraneous_destination' , default = False ) , dest_mode = destmode , mode = mode , overwrite = _merge_setting ( cli_options , conf_options , 'overwrite' , default = True ) , recursive = _merge_setting ( cli_options , conf_options , 'recursive' , default = True ) , rename = _merge_setting ( cli_options , conf_options , 'rename' , default = False ) , ) , skip_on_options = blobxfer . models . options . SkipOn ( filesize_match = _merge_setting ( cli_sod , conf_sod , 'filesize_match' , default = False ) , lmt_ge = _merge_setting ( cli_sod , conf_sod , 'lmt_ge' , default = False ) , md5_match = _merge_setting ( cli_sod , conf_sod , 'md5_match' , default = False ) , ) , ) for src in conf [ 'source' ] : sa = next ( iter ( src ) ) asp = blobxfer . operations . azure . SourcePath ( ) asp . add_path_with_storage_account ( src [ sa ] , sa ) incl = _merge_setting ( cli_conf , conf , 'include' , default = None ) if blobxfer . util . is_not_empty ( incl ) : asp . add_includes ( incl ) excl = _merge_setting ( cli_conf , conf , 'exclude' , default = None ) if blobxfer . util . is_not_empty ( excl ) : asp . add_excludes ( excl ) scs . add_azure_source_path ( asp ) for dst in conf [ 'destination' ] : if len ( dst ) != 1 : raise RuntimeError ( 'invalid number of destination pairs specified per entry' ) sa = next ( iter ( dst ) ) adp = blobxfer . operations . azure . DestinationPath ( ) adp . add_path_with_storage_account ( dst [ sa ] , sa ) scs . add_azure_destination_path ( adp ) specs . append ( scs ) return specs
Create a list of SyncCopy Specification objects from configuration
    :param dict ctx_cli_options: cli options
    :param dict config: config dict
    :rtype: list
    :return: list of SyncCopy Specification objects