Skip to content

dp3.bin.sh

Shell-oriented CLI for interacting with a running DP3 API.

render_completion_shellcode 

render_completion_shellcode(shell_name: str, command_names: list[str]) -> str

Render shell completion registration code using argcomplete.

Source code in dp3/bin/sh.py 
16
17
18
19
def render_completion_shellcode(shell_name: str, command_names: list[str]) -> str:
    """Render shell completion registration code using argcomplete."""
    commands = command_names or ["dp3"]
    return shellcode(commands, shell=shell_name)

handle_completion 

handle_completion(_client, args) -> int

Print shell completion registration code.

Source code in dp3/bin/sh.py 
22
23
24
25
26
27
def handle_completion(_client, args) -> int:
    """Print shell completion registration code."""
    sys.stdout.write(render_completion_shellcode(args.shell_name, args.completion_commands))
    if not sys.stdout.isatty():
        sys.stdout.write("\n")
    return 0

register_completion_parser 

register_completion_parser(commands) -> None

Register shell completion helpers on the root parser.

Source code in dp3/bin/sh.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def register_completion_parser(commands) -> None:
    """Register shell completion helpers on the root parser."""
    completion_parser = commands.add_parser(
        "completion",
        help="Print shell completion scripts.",
        description=(
            "Print shell completion scripts backed by argcomplete. Evaluate the generated "
            "output in your shell or source it from your shell startup file."
        ),
    )
    completion_parser.add_argument(
        "shell_name",
        help="Shell name.",
        choices=["bash", "zsh", "fish", "tcsh", "powershell"],
    )
    completion_parser.add_argument(
        "-c",
        "--command",
        dest="completion_commands",
        action="append",
        default=None,
        help=(
            "Command name to register completion for. Repeat to register multiple "
            "commands. Use 'dp3' for 'dp3 sh' and '<APPNAME>sh' for wrapper commands."
        ),
    )
    completion_parser.set_defaults(
        handler=handle_completion, requires_api=False, load_model_spec=False
    )

init_parser 

init_parser(parser: ArgumentParser) -> None

Initialize the shell-oriented CLI parser.

Source code in dp3/bin/sh.py 
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
def init_parser(parser: argparse.ArgumentParser) -> None:
    """Initialize the shell-oriented CLI parser."""
    config_action = parser.add_argument(
        "-c",
        "--config",
        default=None,
        help=(
            "Path to the DP3 configuration directory. "
            "Resolution order: --config, DP3_CONFIG_DIR, ./config."
        ),
    )
    config_action.completer = DirectoriesCompleter()
    parser.add_argument(
        "-u",
        "--url",
        default=None,
        help="Base URL of the DP3 API. When omitted, localhost defaults are probed.",
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=float,
        default=5.0,
        help="HTTP timeout in seconds.",
    )

    commands = parser.add_subparsers(dest="sh_command", required=True)
    health.register_parser(commands)
    datapoints.register_parser(commands)
    entities.register_parser(commands)
    entity.register_parser(commands)
    control.register_parser(commands)
    telemetry.register_parser(commands)
    register_completion_parser(commands)

run 

run() -> None

Run the shell-oriented CLI as a standalone script.

Source code in dp3/bin/sh.py 
 97
 98
 99
100
101
102
103
def run() -> None:
    """Run the shell-oriented CLI as a standalone script."""
    parser = argparse.ArgumentParser(prog="dp3 sh")
    init_parser(parser)
    argcomplete.autocomplete(parser)
    args = parser.parse_args()
    sys.exit(main(args))

main 

main(args) -> int

Execute a parsed shell-oriented CLI command.

Source code in dp3/bin/sh.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def main(args) -> int:
    """Execute a parsed shell-oriented CLI command."""
    parsed_args = args
    prepare_args = getattr(args, "prepare_args", None)
    if prepare_args is not None:
        parsed_args, exit_code = prepare_args(args)
        if exit_code is not None:
            return exit_code

    config_dir = resolve_config_dir(args.config)
    parsed_args.config_dir = config_dir

    model_spec = None
    if getattr(parsed_args, "load_model_spec", True):
        try:
            config = read_config_dir(config_dir, recursive=True)
            model_spec = ModelSpec(config.get("db_entities"))
        except Exception as e:
            if not getattr(parsed_args, "ignore_config_errors", False):
                print(f"Cannot read config directory '{config_dir}': {e}", file=sys.stderr)
                return 1
    parsed_args.model_spec = model_spec

    if not getattr(parsed_args, "requires_api", True):
        return parsed_args.handler(None, parsed_args)

    try:
        client = DP3APIClient(config_dir, args.url, args.timeout, model_spec=model_spec)
        return parsed_args.handler(client, parsed_args)
    except APIError as e:
        print(str(e), file=sys.stderr)
        return 1