def forcemerge_index()

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

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

def forcemerge_index(task: 'Task', stepname, var: DotMap, **kwargs) -> None:
    """Force merge redacted index"""
    missing_data(stepname, kwargs)
    msg = ''
    fmkwargs = {}
    if 'forcemerge' in task.job.config:
        fmkwargs = task.job.config['forcemerge']
    if 'only_expunge_deletes' in fmkwargs and fmkwargs['only_expunge_deletes']:
        msg = 'only expunging deleted docs'
    else:
        mns = 1  # default value
        if 'max_num_segments' in fmkwargs and isinstance(
            fmkwargs['max_num_segments'], int
        ):
            mns = fmkwargs['max_num_segments']
        msg = f'to {mns} segments per shard'
    drm = f'Forcemerge index {var.redaction_target} {msg}'
    metastep(
        task,
        stepname,
        _meta_forcemerge_index,
        task,
        var,
        include_step=True,
        dry_run_msg=drm,
    )