def run()

in core/standalone.py [0:0]

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

def run(pipe):
    import typer
    from typing_extensions import Annotated

    def _main(
        dry_run: Annotated[bool, typer.Option()] = False,
        log_level: Annotated[str, typer.Option(callback=setup_logging("DEBUG"))] = None,
        pipe_mode: Annotated[
            bool,
            typer.Option(
                "--pipe-mode",
                "-p",
                rich_help_panel="UNIX pipe mode",
                help="Read state from standard input and write state to standard output. This is the default mode when executed in a UNIX pipe.",
            ),
        ] = False,
    ):
        logger = logging.getLogger("elastic.pipes.core")

        if sys.stdin.isatty() and not pipe_mode:
            help_message(pipe)
            sys.exit(1)

        try:
            state = receive_state_from_unix_pipe(pipe.logger, pipe.default)
            pipes = get_pipes(state)
        except Error as e:
            fatal(e)

        configs = [c for n, c in pipes if n == pipe.name]
        config = configs[0] if configs else {}
        with ExitStack() as stack:
            try:
                pipe.run(config, state, dry_run, logger, stack)
            except Error as e:
                pipe.logger.critical(e)
                sys.exit(1)
        send_state_to_unix_pipe(pipe.logger, state)

    typer.run(_main)