def _verify_packaging()

in exec/src/klio_exec/commands/run.py [0:0]

• 34 lines of code
• 8 McCabe index (conditional complexity)

    def _verify_packaging(self):
        pipeline_opts = self.config.pipeline_options
        experiments = pipeline_opts.experiments
        fnapi_enabled = "beam_fn_api" in experiments
        has_setup_file = pipeline_opts.setup_file is not None
        setup_file_exists = has_setup_file and os.path.exists(
            pipeline_opts.setup_file
        )
        has_reqs_file = pipeline_opts.requirements_file is not None
        if fnapi_enabled and any([has_setup_file, has_reqs_file]):
            logging.error(
                "The 'beam_fn_api' experiment may not be enabled while "
                "providing a setup.py file and/or a requirements.txt file."
            )
            raise SystemExit(1)

        if not pipeline_opts.streaming:
            if fnapi_enabled:
                logging.warn(
                    "Support for batch jobs using the 'beam_fn_api' "
                    "experiment is still in development. "
                    "Use with caution."
                )

            if not any([fnapi_enabled, setup_file_exists]):
                if has_reqs_file:
                    logging.warn(
                        "Klio jobs are multi-module. "
                        "Thus, a setup.py file is required "
                        "in addition to a requirements.txt file"
                    )
                else:
                    logging.error(
                        "setup.py file either unspecified or not found."
                    )
                    raise SystemExit(1)