Skip to content

CLI surface

The CLI is the un-comtrade console script — installed alongside the Python package. Five outer commands, 22 sub-subcommands, and five output formats (json, table, csv, markdown, text).

API reference

The full CLI surface is generated from the SDK's docstrings via mkdocstrings. The CLI is a thin wrapper around the Python facade; there is no separate API beyond the Python facade.

cli

Top-level package for the un-comtrade CLI.

The CLI is the first consumer of the public SDK API surface (per 031_PRODUCTION_READINESS.md §9 and docs/IMPLEMENTATION_BASELINE_v1.md §7 "Application").

The foundation phase (C-001) ships:

  • A argparse-based root parser (:func:un_comtrade.cli.main.build_parser).
  • A main(argv) entry point (:func:un_comtrade.cli.main.main).
  • Exit codes (:data:EXIT_OK, ...).
  • Configuration loader (:func:un_comtrade.cli.utils.load_cli_configuration).
  • Output formatters: JSON (functional), TABLE and CSV (placeholders for P7-002+).
  • Zero business commands. The CLI is ready for P7-002 to register metadata, trade, storage, etl, and analytics commands.

Public surface (re-exported for convenience):

  • :func:build_parser
  • :func:main
  • :data:EXIT_OK / :data:EXIT_GENERIC_ERROR / :data:EXIT_USER_ERROR / :data:EXIT_CONFIG_ERROR / :data:EXIT_NETWORK_ERROR / :data:EXIT_AUTH_ERROR
  • :class:CLIError / :class:CLIConfigurationError
  • :func:load_cli_configuration
  • :data:OUTPUT_FORMATS

CLIConfigurationError

Bases: CLIError

Raised when the CLI cannot load or apply a configuration (missing API key, malformed ~/.un_comtrade config, etc.).

Source code in un_comtrade/cli/utils/exceptions.py 
24
25
26
27
28
class CLIConfigurationError(CLIError):
    """Raised when the CLI cannot load or apply
    a configuration (missing API key, malformed
    ``~/.un_comtrade`` config, etc.).
    """

CLIError

Bases: ComtradeError

Base class for CLI-raised errors.

Inherits from :class:ComtradeError so that callers may catch SDK and CLI errors with the same except clause.

Source code in un_comtrade/cli/utils/exceptions.py 
15
16
17
18
19
20
21
class CLIError(ComtradeError):
    """Base class for CLI-raised errors.

    Inherits from :class:`ComtradeError` so that
    callers may catch SDK and CLI errors with the
    same ``except`` clause.
    """

ProgressReporter

Minimal progress reporter.

Output shape::

[trade/exports] page 1: 250 records (total 250)

The reporter writes to stderr (not stdout) so that the data stream on stdout remains machine-readable when --output-format json is used.

Source code in un_comtrade/cli/utils/progress.py 
20
21
22
23
24
25
26
27
28
29
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
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
class ProgressReporter:
    """Minimal progress reporter.

    Output shape::

        [trade/exports] page 1: 250 records (total 250)

    The reporter writes to stderr (not stdout) so
    that the data stream on stdout remains
    machine-readable when ``--output-format
    json`` is used.
    """

    def __init__(
        self,
        *,
        label: str,
        enabled: bool,
        stream=None,
        force: bool = False,
    ) -> None:
        self._label = label
        self._enabled = enabled
        self._stream = stream if stream is not None else sys.stderr
        # ``force=True`` bypasses the TTY check so
        # tests can capture progress output
        # regardless of pytest's stream state.
        self._force = force
        self._total = 0

    def update(self, current_total: int, *, page: int | None = None) -> None:
        """Update the running total and (optionally)
        the page number. Writes a single line to
        the configured stream.
        """
        self._total = current_total
        if not self._should_write():
            return
        suffix = (
            f" page {page}: "
            if page is not None
            else " "
        )
        self._stream.write(
            f"[{self._label}]{suffix}{current_total} records\n"
        )
        self._stream.flush()

    def finish(self, total: int) -> None:
        """Final summary line.
        """
        if not self._should_write():
            return
        self._stream.write(
            f"[{self._label}] done: {total} records\n"
        )
        self._stream.flush()

    def _should_write(self) -> bool:
        if not self._enabled:
            return False
        if self._force:
            return True
        stream = self._stream
        # ``isatty`` is on the IO base class; some
        # wrappers (e.g. pytest's capture) return
        # False. In that case we honour the user's
        # ``enabled`` flag rather than guessing.
        return bool(getattr(stream, "isatty", lambda: False)())

update 

update(
    current_total: int, *, page: int | None = None
) -> None

Update the running total and (optionally) the page number. Writes a single line to the configured stream.

Source code in un_comtrade/cli/utils/progress.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def update(self, current_total: int, *, page: int | None = None) -> None:
    """Update the running total and (optionally)
    the page number. Writes a single line to
    the configured stream.
    """
    self._total = current_total
    if not self._should_write():
        return
    suffix = (
        f" page {page}: "
        if page is not None
        else " "
    )
    self._stream.write(
        f"[{self._label}]{suffix}{current_total} records\n"
    )
    self._stream.flush()

finish 

finish(total: int) -> None

Final summary line.

Source code in un_comtrade/cli/utils/progress.py 
68
69
70
71
72
73
74
75
76
def finish(self, total: int) -> None:
    """Final summary line.
    """
    if not self._should_write():
        return
    self._stream.write(
        f"[{self._label}] done: {total} records\n"
    )
    self._stream.flush()

build_parser 

build_parser(
    *,
    prog: str = "un-comtrade",
    description: str = "Command-line interface for the un-comtrade-sdk. The CLI is a thin consumer of the SDK's public API surface; the SDK itself is fully usable from Python.",
) -> ArgumentParser

Construct the root argparse parser.

Returns a parser with global options and subparsers for every registered command. Tests use this to construct the parser without running :func:main.

Source code in un_comtrade/cli/main.py 
105
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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def build_parser(
    *,
    prog: str = "un-comtrade",
    description: str = (
        "Command-line interface for the un-comtrade-sdk. "
        "The CLI is a thin consumer of the SDK's public "
        "API surface; the SDK itself is fully usable "
        "from Python."
    ),
) -> argparse.ArgumentParser:
    """Construct the root argparse parser.

    Returns a parser with global options and
    subparsers for every registered command. Tests
    use this to construct the parser without
    running :func:`main`.
    """
    parser = argparse.ArgumentParser(
        prog=prog,
        description=description,
        add_help=True,
    )

    parser.add_argument(
        "--version",
        action="version",
        version=f"un-comtrade {un_comtrade.__version__} "
        f"(un-comtrade-sdk {un_comtrade.__version__})",
    )

    # ----- Global options ----------------------------------------------

    parser.add_argument(
        "--api-key",
        dest="api_key",
        default=None,
        help=(
            "Override the UN_COMTRADE_KEY env var. "
            "Useful when running one-shot commands "
            "without persisting the key in your "
            "environment."
        ),
    )
    parser.add_argument(
        "--log-level",
        dest="log_level",
        default=None,
        help=(
            "Override the configured log level. "
            "One of DEBUG / INFO / WARNING / "
            "ERROR / CRITICAL."
        ),
    )
    parser.add_argument(
        "--output-format",
        dest="output_format",
        default=None,
        choices=OUTPUT_FORMATS,
        help=(
            "Render command output in the chosen "
            f"format. Default: json. Choices: "
            f"{', '.join(OUTPUT_FORMATS)}."
        ),
    )
    parser.add_argument(
        "--output",
        dest="output",
        default=None,
        help=(
            "Write the rendered output to PATH "
            "instead of stdout. Useful for "
            "redirecting into scripts / pipes."
        ),
    )

    # ----- Subparsers --------------------------------------------------

    names = known_command_names()
    if names:
        sub = parser.add_subparsers(
            title="commands",
            dest="command",
            metavar="<command>",
            help="Available subcommands.",
        )
        for name in names:
            cmd = get_command(name)
            if cmd is None:
                continue
            # Group commands (``metadata`` and
            # future siblings) expose
            # ``install_subparser`` and own their
            # own sub-subparsers. Single-shot
            # commands get a flat subparser that
            # forwards all args to ``__call__``.
            install = getattr(cmd, "install_subparser", None)
            if callable(install):
                install(sub)
            else:
                sub.add_parser(
                    name,
                    help=cmd.help,
                    description=cmd.help,
                    add_help=True,
                )

    return parser

load_cli_configuration 

load_cli_configuration(
    *,
    api_key: str | None = None,
    log_level: str | None = None,
    base_url: str | None = None,
    timeout: float | None = None,
) -> Configuration

Load the SDK :class:Configuration from the environment, applying CLI-side overrides on top.

Parameters

api_key Override for the subscription key. When supplied, this takes precedence over the UN_COMTRADE_KEY env var. log_level Override for the log level. When supplied, this takes precedence over the UN_COMTRADE_LOG_LEVEL env var. base_url Override for the upstream base URL. timeout Override for the per-request timeout (seconds).

Returns

Configuration A frozen :class:Configuration with all overrides applied.

Raises

CLIConfigurationError When an override value is invalid (unknown log level, malformed URL, etc.).

Source code in un_comtrade/cli/utils/config_loader.py 
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def load_cli_configuration(
    *,
    api_key: str | None = None,
    log_level: str | None = None,
    base_url: str | None = None,
    timeout: float | None = None,
) -> Configuration:
    """Load the SDK :class:`Configuration` from
    the environment, applying CLI-side overrides
    on top.

    Parameters
    ----------
    api_key
        Override for the subscription key. When
        supplied, this takes precedence over the
        ``UN_COMTRADE_KEY`` env var.
    log_level
        Override for the log level. When supplied,
        this takes precedence over the
        ``UN_COMTRADE_LOG_LEVEL`` env var.
    base_url
        Override for the upstream base URL.
    timeout
        Override for the per-request timeout
        (seconds).

    Returns
    -------
    Configuration
        A frozen :class:`Configuration` with all
        overrides applied.

    Raises
    ------
    CLIConfigurationError
        When an override value is invalid (unknown
        log level, malformed URL, etc.).
    """
    cfg = load_configuration()

    # Apply overrides by reconstructing the
    # Configuration via `dataclasses.replace`. The
    # Configuration is frozen, so we cannot
    # mutate it in place.
    from dataclasses import replace
    overrides: dict[str, Any] = {}
    if api_key is not None:
        overrides["api_key"] = api_key
    if log_level is not None:
        overrides["log_level"] = _validate_log_level(log_level)
    if base_url is not None:
        overrides["base_url"] = base_url
    if timeout is not None:
        if timeout <= 0:
            raise CLIConfigurationError(
                f"timeout must be positive; got {timeout}"
            )
        overrides["timeout_seconds"] = timeout
    if overrides:
        cfg = replace(cfg, **overrides)
    return cfg

load_dataset 

load_dataset(path: str | Path) -> Tuple

Load a :class:CanonicalDataset from a previously-stored file via the public Storage API.

Parameters

path Path to a stored dataset. The extension determines the backend.

Returns

CanonicalDataset The deserialised dataset.

Raises

CLIConfigurationError When the path does not exist, the extension is unsupported, or the backend cannot read the file.

Source code in un_comtrade/cli/utils/dataset_loader.py 
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def load_dataset(path: str | Path) -> Tuple:
    """Load a :class:`CanonicalDataset` from a
    previously-stored file via the public Storage
    API.

    Parameters
    ----------
    path
        Path to a stored dataset. The extension
        determines the backend.

    Returns
    -------
    CanonicalDataset
        The deserialised dataset.

    Raises
    ------
    CLIConfigurationError
        When the path does not exist, the
        extension is unsupported, or the backend
        cannot read the file.
    """
    p = Path(path)
    if not p.exists():
        raise CLIConfigurationError(
            f"dataset file does not exist: {p}"
        )
    backend = _detect_backend(p)
    storage = StorageRegistry().get(backend)
    if storage is None:
        raise CLIConfigurationError(
            f"storage backend {backend.name!r} is not "
            f"available; install the optional "
            f"dependency (e.g. `pip install "
            f"un-comtrade-sdk[{backend.name.lower()}]`)"
        )
    config = StorageConfig(root=str(p))
    try:
        return storage.read(config)
    except Exception as exc:
        # The Storage layer raises a variety of
        # exception types (StorageError,
        # OSError, ...); normalise them.
        raise CLIConfigurationError(
            f"failed to read dataset {p}: {exc}"
        ) from exc

make_progress_reporter 

make_progress_reporter(
    *,
    label: str,
    enabled: bool,
    force: bool = False,
    stream=None,
) -> ProgressReporter

Factory that returns either a real :class:ProgressReporter or a no-op stub.

Always returns a reporter object so callers can invoke .update(...) unconditionally.

Source code in un_comtrade/cli/utils/progress.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def make_progress_reporter(
    *,
    label: str,
    enabled: bool,
    force: bool = False,
    stream=None,
) -> ProgressReporter:
    """Factory that returns either a real
    :class:`ProgressReporter` or a no-op stub.

    Always returns a reporter object so callers
    can invoke ``.update(...)`` unconditionally.
    """
    if not enabled and not force:
        return _NullReporter(label)
    return ProgressReporter(
        label=label,
        enabled=enabled,
        force=force,
        stream=stream,
    )

render_to_destination 

render_to_destination(
    text: str, *, output: str | None
) -> None

Write text to output (a file path) or to stdout when output is None.

Raises

CLIConfigurationError When output cannot be opened for writing (permission, missing parent dir, etc.).

Source code in un_comtrade/cli/utils/output.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def render_to_destination(
    text: str,
    *,
    output: str | None,
) -> None:
    """Write ``text`` to ``output`` (a file path)
    or to ``stdout`` when ``output`` is ``None``.

    Raises
    ------
    CLIConfigurationError
        When ``output`` cannot be opened for
        writing (permission, missing parent dir,
        etc.).
    """
    if output is None:
        sys.stdout.write(text)
        return
    path = Path(output)
    try:
        # ``newline=""`` keeps csv / json content
        # byte-identical (no extra newline on
        # Windows).
        with path.open("w", encoding="utf-8", newline="") as f:
            f.write(text)
    except OSError as exc:
        raise CLIConfigurationError(
            f"cannot write output file {output}: {exc}"
        ) from exc

Examples

un-comtrade trade exports --reporter 699 --period 2022 --partner 0 --output-format markdown
  • RECIPE-091 — Drive metadata commands from the CLI.
  • RECIPE-095 — Drive trade commands from the CLI.
  • RECIPE-099 — Drive analytics commands from the CLI.