def main()

in core/import.py [0:0]

• 24 lines of code
• 12 McCabe index (conditional complexity)

def main(ctx: Ctx, log: Logger, dry_run: bool):
    """Import data from file or standard input."""

    format = ctx.format
    if format is None:
        if ctx.file_name:
            format = Path(ctx.file_name).suffix.lower()[1:]
            log.debug(f"import file format guessed from file extension: {format}")
        else:
            format = "yaml"
            log.debug(f"assuming import file format: {format}")

    if not ctx.file_name and sys.stdin.isatty() and not ctx.interactive:
        fatal("To use `elastic.pipes.core.import` interactively, set `interactive: true` in its configuration.")

    if dry_run:
        return

    node = ctx.get_binding("state").node
    msg_state = "everything" if node is None else f"'{node}'"
    msg_file_name = f"'{ctx.file_name}'" if ctx.file_name else "standard input"
    log.info(f"importing {msg_state} from {msg_file_name}...")

    if ctx.file_name:
        with open(Path(ctx.base_dir) / ctx.file_name, "r") as f:
            warn_interactive(f)
            ctx.state = deserialize(f, format=format) or {}
    else:
        warn_interactive(sys.stdin)
        ctx.state = deserialize(sys.stdin, format=format) or {}