def _get_batch_user_input_job_context()

in cli/src/klio_cli/commands/job/create.py [0:0]

• 45 lines of code
• 5 McCabe index (conditional complexity)

    def _get_batch_user_input_job_context(self, kwargs):
        default_batch_event_input = "{}_input_ids.txt".format(
            kwargs.get("job_name")
        )
        batch_event_input = kwargs.get("batch_event_input")
        if not batch_event_input:
            batch_event_input = click.prompt(
                "Batch event input file", default=default_batch_event_input
            )

        default_batch_data_input = "{}-input".format(kwargs.get("job_name"))
        batch_data_input = kwargs.get("batch_data_input")
        if not batch_data_input:
            batch_data_input = click.prompt(
                "Batch data input directory", default=default_batch_data_input
            )

        default_batch_event_output = "{}_output_elements".format(
            kwargs.get("job_name")
        )
        batch_event_output = kwargs.get("batch_event_output")
        if not batch_event_output:
            batch_event_output = click.prompt(
                "Batch event output file", default=default_batch_event_output
            )

        default_batch_data_output = "{}-output".format(kwargs.get("job_name"))
        batch_data_output = kwargs.get("batch_data_output")
        if not batch_data_output:
            batch_data_output = click.prompt(
                "Batch data output directory",
                default=default_batch_data_output,
            )

        job_context = {
            "inputs": [
                {
                    "event_location": batch_event_input,
                    "data_location": batch_data_input,
                }
            ],
            "outputs": [
                {
                    "event_location": batch_event_output,
                    "data_location": batch_data_output,
                }
            ],
        }
        return job_context