def fix_aliases()

in src/es_pii_tool/helpers/steps.py [0:0]

• 41 lines of code
• 6 McCabe index (conditional complexity)

def fix_aliases(task: 'Task', stepname, var: DotMap, **kwargs) -> None:
    """Using the aliases collected from var.index, update mount_name and verify"""
    missing_data(stepname, kwargs)
    data = kwargs['data']
    step = Step(task=task, stepname=stepname)
    if step.finished():
        logger.info('%s: already completed', step.stub)
        return
    step.begin()
    if data.data_stream:
        msg = 'Cannot apply aliases to indices in data_stream'
        logger.debug(msg)
        step.add_log(msg)
        step.end(completed=True, errors=False, logmsg=f'{stepname} completed')
        return
    alias_names = var.aliases.toDict().keys()
    if not alias_names:
        msg = f'No aliases associated with index {var.index}'
        step.add_log(msg)
        logger.info(msg)
    elif not task.job.dry_run:
        msg = f'Transferring aliases to new index ' f'{var.mount_name}'
        logger.debug(msg)
        step.add_log(msg)
        var.client.indices.update_aliases(
            actions=get_alias_actions(var.index, var.mount_name, var.aliases.toDict())
        )
        verify = var.client.indices.get(index=var.mount_name)[var.mount_name][
            'aliases'
        ].keys()
        if alias_names != verify:
            msg = f'Alias names do not match! {alias_names} does not match: {verify}'
            logger.critical(msg)
            step.add_log(msg)
            failed_step(
                task, step, ValueMismatch(msg, 'alias names mismatch', alias_names)
            )
    else:
        msg = 'Dry-Run: alias transfer not executed'
        logger.debug(msg)
        step.add_log(msg)
    step.end(completed=True, errors=False, logmsg=f'{stepname} completed')