GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — unstract-cli/src/unstract_cli/app.py at main · Zipstack/unstract-cli

Expand file tree

/

app.py

Copy path
More file actions
More file actions

Latest commit

 

History

History
History
321 lines (277 loc) · 10.7 KB
/

app.py

Copy path

File metadata and controls

321 lines (277 loc) · 10.7 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
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
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
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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
"""The root Click application: global options and the command groups.
Global options are declared once here and reach every command through the Click
context, so no command re-implements profile selection or output formatting.
"""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any
import click
from unstract_cli.commands.config_cmd import config_group
from unstract_cli.config import (
DOCSTUDIO,
LLMWHISPERER,
SECRET_SETTINGS,
ConfigError,
ResolvedConfig,
load_config,
set_config_path,
)
from unstract_cli.core.clients import DEFAULT_TRANSPORT_TIMEOUT
from unstract_cli.core.discover import TIERS, discover
from unstract_cli.core.errors import CLIError, ExitCode, set_warning_sink
from unstract_cli.core.output import (
AgentMode,
OutputFormat,
diagnostic,
emit_result,
resolve_format,
)
@dataclass
class Context:
"""Everything a command needs from the global options."""
output: OutputFormat = OutputFormat.TABLE
quiet: bool = False
verbosity: int = 0
profile: str | None = None
#: Transport timeout for the clients that have none of their own.
transport_timeout: float | None = DEFAULT_TRANSPORT_TIMEOUT
#: Command-line overrides, keyed `product.setting` -- the top tier of
#: flag > env > profile > default.
overrides: dict[str, Any] = field(default_factory=dict)
_config: ResolvedConfig | None = field(default=None, repr=False)
@property
def config(self) -> ResolvedConfig:
"""Load the config lazily, so commands that need none never read a file."""
if self._config is None:
try:
cfg = load_config()
except ConfigError as exc:
raise CLIError(str(exc), ExitCode.USAGE) from exc
for warning in cfg.warnings:
diagnostic(warning, quiet=self.quiet, verbosity=self.verbosity)
self._config = ResolvedConfig(
file=cfg, profile_name=self.profile, overrides=self.overrides
)
return self._config
def override(self, product: str, values: dict[str, Any]) -> None:
"""Record the connection flags given for one product.
Called from the product group, before any command runs, so the flag tier
is populated by the time a command resolves anything.
"""
for key, value in values.items():
if value is None:
continue
if key in SECRET_SETTINGS:
diagnostic(
"warning: a key passed on the command line lands in shell "
"history and in the process list. Prefer the environment "
"variable or `env:` indirection in a profile.",
quiet=self.quiet,
verbosity=self.verbosity,
)
self.overrides[f"{product}.{key}"] = value
def secrets(self) -> list[str]:
"""Resolved credentials, for scrubbing anything on its way to a stream."""
out: list[str] = []
for product, key in (
(LLMWHISPERER, "api_key"),
(DOCSTUDIO, "api_key"),
(DOCSTUDIO, "platform_key"),
):
try:
if value := self.config.get(product, key):
out.append(str(value))
except (ConfigError, CLIError):
# Unresolvable means unprintable too, and raising would
# replace a finished report with a config error.
continue
return out
pass_context = click.make_pass_decorator(Context, ensure=True)
# `invoke_without_command` so `--discover` answers on its own: it is how a
# caller learns which commands exist.
@click.group(
invoke_without_command=True,
context_settings={"help_option_names": ["-h", "--help"]},
)
@click.option(
"--config",
"config_file",
default=None,
type=click.Path(dir_okay=False),
help="Config file to use, overriding discovery.",
)
@click.option("--profile", "-p", default=None, help="Configuration profile to use.")
@click.option(
"--output",
"-o",
default=None,
type=click.Choice([f.value for f in OutputFormat]),
help="Output format. Defaults to table, or to json when --agent resolves "
"to yes; pass it explicitly to parse the output.",
)
@click.option(
"--agent",
type=click.Choice([m.value for m in AgentMode]),
default=AgentMode.AUTO.value,
help="Whether a coding agent is driving this: sets the default format to "
"json. Only the default -- an explicit --output always wins.",
)
@click.option(
"--quiet",
"-q",
is_flag=True,
default=False,
help="Suppress diagnostics on stderr. stdout is unaffected.",
)
@click.option("--verbose", "-v", count=True, help="Increase diagnostic detail.")
@click.option(
"--discover",
"discover_tier",
type=click.Choice(TIERS),
default=None,
help="Describe this CLI as JSON instead of running a command, useful for agents.",
)
@click.version_option(package_name="unstract-cli")
@click.pass_context
def cli(
ctx: click.Context,
config_file: str | None,
profile: str | None,
output: str | None,
agent: str,
quiet: bool,
verbose: int,
discover_tier: str | None,
) -> None:
"""The official CLI for Unstract.
LLMWhisperer extracts text and layout from documents; Document Studio runs
them through API deployments that return structured JSON.
Scripting or driving this from an agent: `-o json` prints one
`{ok, data, error, meta}` envelope on stdout and nothing else, failures
exit non-zero with a stable code, and `--discover groups|summary|full`
describes the commands, their flags and the output contract as JSON without
running anything.
"""
set_config_path(config_file)
# Filled in rather than replaced: the entry point holds this object, so a
# failure anywhere below renders in the format resolved here.
obj = ctx.ensure_object(Context)
obj.output = resolve_format(output, agent)
obj.quiet = quiet
obj.verbosity = verbose
obj.profile = profile
# Modules the output layer imports cannot import it back, so their notes
# reach it through this sink rather than stderr unfiltered.
set_warning_sink(
lambda message: diagnostic(message, quiet=obj.quiet, verbosity=obj.verbosity)
)
if discover_tier:
# Discovery has to answer before any configuration exists.
emit_result(discover(cli, discover_tier), OutputFormat.JSON)
ctx.exit(int(ExitCode.SUCCESS))
if ctx.invoked_subcommand is None:
if obj.output is not OutputFormat.TABLE:
# stdout carries one envelope and nothing else: help printed there
# with exit 0 tells a parser that work it never did succeeded.
raise CLIError(
"No command given.",
ExitCode.USAGE,
hint="`--discover groups` lists what can be run, as JSON.",
)
click.echo(ctx.get_help())
ctx.exit(int(ExitCode.SUCCESS))
#: The connection flags a product group can carry, named after the setting
#: each one overrides.
_CONNECTION_FLAGS: dict[str, str] = {
"base_url": "Service URL to use.",
"api_key": "API key to use.",
"org_id": "Organisation to run against.",
"platform_key": "Platform key to use, for the commands that take one.",
}
def _connection_options(*settings: str) -> Callable[[Any], Any]:
"""The per-product connection settings, as flags.
They sit on the product group rather than on each command: they say where to
connect, which is the same question for every command underneath.
"""
options = [
click.option(
f"--{name.replace('_', '-')}", default=None, help=_CONNECTION_FLAGS[name]
)
for name in ("base_url", *settings)
]
def decorate(func: Any) -> Any:
for option in reversed(options):
func = option(func)
return func
return decorate
@cli.group("whisper")
@_connection_options("api_key")
@pass_context
def whisper_group(ctx: Context, **overrides: str | None) -> None:
"""Extract text and layout from documents with LLMWhisperer."""
ctx.override(LLMWHISPERER, overrides)
@cli.group("docstudio")
@_connection_options("api_key", "org_id", "platform_key")
@click.option(
"--transport-timeout",
type=click.FloatRange(min=0),
default=DEFAULT_TRANSPORT_TIMEOUT,
show_default=True,
help="Seconds before a stalled connection is given up on. 0 waits forever.",
)
@pass_context
def docstudio_group(
ctx: Context, transport_timeout: float, **overrides: str | None
) -> None:
"""Run Document Studio API deployments."""
ctx.transport_timeout = transport_timeout or None
ctx.override(DOCSTUDIO, overrides)
@docstudio_group.group("deployment")
def deployment_group() -> None:
"""Work with a deployed API."""
@cli.group("auth")
@_connection_options("platform_key")
@click.option(
"--transport-timeout",
type=click.FloatRange(min=0),
default=DEFAULT_TRANSPORT_TIMEOUT,
show_default=True,
help="Seconds before a stalled connection is given up on. 0 waits forever.",
)
@pass_context
def auth_group(
ctx: Context, transport_timeout: float | None, **overrides: str | None
) -> None:
"""Sign in, and identify the credential you are using.
Its flags configure the platform key, which is the credential that knows
which organisation it belongs to. A deployment key does not: it authenticates
against the deployment it was minted for and never reaches this endpoint.
"""
# The same spelling as every other group: the clients are given no bound
# of their own, so an unset flag would wait on a black-holed host forever.
ctx.transport_timeout = transport_timeout or None
ctx.override(DOCSTUDIO, overrides)
cli.add_command(config_group)
# Imported for the side effect of registering commands, and last because they
# hang those commands off the groups declared above.
from unstract_cli.commands import ( # noqa: E402,F401
clone_cmd,
docstudio_cmd,
platform_cmd,
whisper_cmd,
)
def command_tree() -> dict[str, Any]:
"""The registered command tree, read back from Click itself.
Describing commands anywhere but from the parser lets the description drift
from what the parser accepts, so discovery and help always read this.
"""
def walk(command: click.Command) -> dict[str, Any]:
entry: dict[str, Any] = {"help": (command.help or "").strip().split("\n")[0]}
if isinstance(command, click.Group):
entry["commands"] = {
name: walk(sub) for name, sub in sorted(command.commands.items())
}
return entry
return walk(cli)["commands"]
__all__ = ["Context", "cli", "command_tree", "pass_context"]
You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.