Skip to content

main ¤

The main CLI function.

Functions:

  • main

    Run the main program.

main ¤

main(args: list[str] | None = None) -> int

Run the main program.

This function is executed when you type aria2p or python -m aria2p.

Parameters:

  • args (list[str] | None, default: None ) –

    Parameters passed from the command line.

Returns:

  • int

    An exit code.

Source code in src/aria2p/cli/main.py
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def main(args: list[str] | None = None) -> int:
    """Run the main program.

    This function is executed when you type `aria2p` or `python -m aria2p`.

    Parameters:
        args: Parameters passed from the command line.

    Returns:
        An exit code.
    """
    parser = get_parser()
    opts = parser.parse_args(args=args)
    kwargs = opts.__dict__

    log_level = kwargs.pop("log_level")
    log_path = kwargs.pop("log_path")

    if log_path:
        log_path = Path(log_path)
        if log_path.is_dir():
            log_path = log_path / "aria2p-{time}.log"
        enable_logger(sink=log_path, level=log_level or "WARNING")
    elif log_level:
        enable_logger(sink=sys.stderr, level=log_level)

    logger.debug("Checking arguments")
    check_args(parser, opts)

    logger.debug("Instantiating API")
    api = API(
        Client(
            host=kwargs.pop("host"),
            port=kwargs.pop("port"),
            secret=kwargs.pop("secret"),
            timeout=kwargs.pop("client_timeout"),
        ),
    )

    logger.info(f"API instantiated: {api!r}")

    # Warn if no aria2 daemon process seems to be running
    logger.debug("Testing connection")
    try:
        api.client.get_version()
    except requests.ConnectionError as error:
        print(f"[ERROR] {error}", file=sys.stderr)
        print(file=sys.stderr)
        print("Please make sure that an instance of aria2c is running with RPC mode enabled,", file=sys.stderr)
        print("and that you have provided the right host, port and secret token.", file=sys.stderr)
        print("More information at https://pawamoy.github.io/aria2p.", file=sys.stderr)
        return 2

    subcommand = kwargs.pop("subcommand")
    kwargs.pop("debug_info")

    if subcommand:
        logger.debug("Running subcommand " + subcommand)
    try:
        return commands[subcommand](api, **kwargs)  # type: ignore
    except ClientException as error:
        print(str(error), file=sys.stderr)
        return error.code