API Reference¶
Auto-generated from the clickwork source tree's docstrings. This page is the authoritative list of public symbols — anything not rendered here is either private or not part of the 1.0 semver surface. See API Policy for the compatibility promise.
clickwork¶
clickwork: Reusable CLI framework for project automation.
This package provides the building blocks for project-specific CLI tools. It handles plugin discovery, layered config, subprocess management, and common utilities so command authors can focus on business logic.
Public API
create_cli - Build a CLI with global flags and plugin discovery add_global_option - Install a Click option at root + every group + every subcommand load_config - Load layered TOML config (for custom config scenarios) CliContext - Typed context object passed to every command pass_cli_context - Decorator for commands (handles nested group footgun) Secret - Redacted wrapper for sensitive config values CliProcessError - Exception raised when subprocess fails PrerequisiteError - Exception raised when a required tool is missing ConfigError - Exception raised when config validation fails HttpError - Exception raised when an HTTP call returns non-2xx ClickworkDiscoveryError - Exception raised when strict discovery finds a broken import, missing cli attribute, or similar command-discovery failure (opt-in via create_cli(..., strict=True)) platform_dispatch - Decorator that routes a command to a per-OS impl platform - Submodule exposing dispatch(), is_linux/macos/windows http - Submodule exposing get/post/put/delete + HttpError testing - Submodule exposing run_cli() + make_test_cli() helpers
CliContext
dataclass
¶
Shared runtime state threaded through every Click command via ctx.obj.
CliContext flows from the top-level Click group down to every sub-command. It carries: - Parsed flags (dry_run, verbose, quiet, yes) -- so sub-commands don't need to re-declare the same Click options. - A resolved config dict -- populated from layered config files by the CLI harness before any command handler runs. - A bound logger -- so commands log consistently with the same formatter and level set by the top-level --verbose / --quiet flags. - Six optional callable fields (run, capture, ...) -- injected by the CLI harness to abstract over dry-run / subprocess semantics. They default to None so the dataclass can be constructed cheaply in tests without a full harness setup.
The callable fields use repr=False, compare=False because: - repr=False: lambdas have unhelpful repr strings that would clutter debug output; the dataclass repr is already redacted via logger repr. - compare=False: two contexts with identical config but different callable bindings should compare equal for assertion purposes.
Usage (CLI harness): @click.pass_context def cli(ctx: click.Context, dry_run: bool, verbose: int) -> None: ctx.ensure_object(dict) ctx.obj = CliContext( config=load_config(), dry_run=dry_run, verbose=verbose, ... )
Usage (command): @click.pass_obj def deploy(ctx: CliContext) -> None: if ctx.dry_run: ctx.logger.info("[dry-run] would deploy") else: ctx.run(["kubectl", "apply", "-f", "manifests/"])
capture = field(default=None, repr=False, compare=False)
class-attribute
instance-attribute
¶
Run a command and return stripped stdout as a string.
config = field(default_factory=dict)
class-attribute
instance-attribute
¶
Merged config from layered TOML files, keyed by string.
confirm = field(default=None, repr=False, compare=False)
class-attribute
instance-attribute
¶
Prompt the user for confirmation unless --yes is set.
confirm_destructive = field(default=None, repr=False, compare=False)
class-attribute
instance-attribute
¶
Like confirm, but adds extra warnings for irreversible operations.
dry_run = False
class-attribute
instance-attribute
¶
When True, commands should describe actions but not execute them.
env = None
class-attribute
instance-attribute
¶
Active deployment environment (e.g. "staging", "prod"), or None.
logger = field(default_factory=(lambda: logging.getLogger('clickwork')))
class-attribute
instance-attribute
¶
Configured logger instance.
quiet = False
class-attribute
instance-attribute
¶
When True, suppress all output except errors.
require = field(default=None, repr=False, compare=False)
class-attribute
instance-attribute
¶
Assert that a binary exists on PATH, raising PrerequisiteError if not.
Tests can intercept this helper by patching clickwork.prereqs.require
(the public symbol). The CLI harness binds ctx.require through a
module-level wrapper function that re-reads the module attribute on
every call, so unittest.mock.patch("clickwork.prereqs.require")
transparently takes effect. Prior to issue #8's fix, tests had to
reach for the internal clickwork.cli._require alias; that alias
no longer exists -- patch clickwork.prereqs.require instead.
run = field(default=None, repr=False, compare=False)
class-attribute
instance-attribute
¶
Run a command, inheriting stdio. Raises CliProcessError on failure.
run_with_confirm = field(default=None, repr=False, compare=False)
class-attribute
instance-attribute
¶
Confirm then run: wraps confirm() + run() in a single call.
run_with_secrets = field(default=None, repr=False, compare=False)
class-attribute
instance-attribute
¶
Run a subprocess with secrets delivered via env (and optionally stdin).
The helper rejects Secret instances appearing directly in cmd
(argv is visible in ps), places each entry of secrets into the
subprocess's environment, and -- when stdin_secret="NAME" is set
-- additionally pipes secrets["NAME"].get() through the child's
stdin (Wave 1's stdin_text= path). Log lines show env-var NAMES
but redact VALUES. See :func:clickwork.process.run_with_secrets for
the full contract and examples.
verbose = 0
class-attribute
instance-attribute
¶
Verbosity level: 0=normal, 1=-v, 2=-vv, etc.
yes = False
class-attribute
instance-attribute
¶
When True, skip interactive confirmation prompts.
CliProcessError
¶
Bases: Exception
A subprocess failure wrapped with enough context to act on.
subprocess.CalledProcessError carries the raw data but its str is
terse and its fields are loosely typed. This wrapper:
- Exposes returncode, cmd, and stderr as first-class attributes.
- Produces a human-readable message that is immediately actionable in
logs and pytest output without needing to inspect the cause chain.
- Remains a plain Exception subclass so callers can raise / except it
using standard Python idioms.
Usage
try: subprocess.run(["git", "push"], check=True, capture_output=True, text=True) except subprocess.CalledProcessError as e: raise CliProcessError(e) from e
__init__(cause)
¶
Wrap a CalledProcessError with a human-readable message.
Extracts the fields callers care about (returncode, cmd, stderr) and composes a single actionable message so that str(err) immediately shows what failed and why, without needing to unwrap the cause chain.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
cause
|
CalledProcessError
|
The original subprocess.CalledProcessError to wrap. |
required |
ClickworkDiscoveryError
¶
Bases: Exception
Raised when discovery fails under strict=True.
Carries a .failures list of structured DiscoveryFailure records so
a single run can surface every problem it found, not just the first. The
convenience .cause_path attribute points at the first failure's file
(useful for test assertions and one-line error logging); inspect
.failures directly for the full picture.
WHY an exception type of our own rather than reusing click.UsageError
or ImportError: these are classification bugs at the CLI-wiring
layer, not user input mistakes or plain import problems. Giving them a
dedicated type lets consumers catch only this class without accidentally
swallowing unrelated errors, and lets our own except clauses in the
CLI startup path treat strict-discovery failures distinctly from
runtime user errors.
__init__(failures)
¶
Build a discovery error that aggregates one or more failure records.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
failures
|
list[DiscoveryFailure]
|
Non-empty list of structured failure records. The
constructor formats a combined message for |
required |
ConfigError
¶
Bases: Exception
Raised when config validation fails.
This is a user-facing error -- the message should be actionable, telling them which key is missing/invalid and where to fix it.
HttpError
¶
Bases: Exception
Raised when the server returns a non-2xx response.
Mirrors the :class:~clickwork._types.CliProcessError pattern: raw
fields are exposed as structured attributes so callers can branch on
status_code without parsing a string message.
Attributes:
| Name | Type | Description |
|---|---|---|
status_code |
int
|
HTTP status code reported by the server (e.g. 404, 500). |
response_body |
JSONValue | bytes
|
Parsed JSON if the original request was issued with
|
headers |
dict[str, str]
|
Response headers as a plain |
url |
str
|
The URL that produced this error -- useful for logs when a single caller issues multiple requests in a loop. |
Usage::
try:
http.get(url, allowed_hosts=[...])
except HttpError as e:
if e.status_code == 404:
return None # "not found" is expected here
raise # anything else is a real error
__init__(status_code, response_body, headers, url, message)
¶
Build an HttpError with the full context needed for triage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
status_code
|
int
|
HTTP status from the server. |
required |
response_body
|
JSONValue | bytes
|
Parsed JSON body (when the caller opted into
|
required |
headers
|
dict[str, str]
|
Response headers as a plain string-keyed dict. |
required |
url
|
str
|
The URL that produced this error. |
required |
message
|
str
|
A pre-composed human-readable message for |
required |
PrerequisiteError
¶
Bases: Exception
Raised when a required tool is missing or not authenticated.
Commands call ctx.require("docker") at the top of their function body. If the binary is missing or not authenticated, require() raises this exception. The framework's error handler catches it and exits with code 1 (user error) -- the same as CliProcessError.
Raising instead of sys.exit() lets callers catch and recover if they want to (e.g., fall back to an alternative tool), and keeps require() composable and testable.
Secret
¶
An opaque wrapper that prevents accidental logging of sensitive values.
Design goals
- str() / repr() / f-strings always emit "***" so log statements that format a CliContext never leak credentials.
- .get() is the single explicit escape hatch -- its name signals intent at call sites and makes grep-audits straightforward.
- slots removes dict entirely so vars(s) raises TypeError and iterating over an object's attributes cannot surface the value.
- Pickling is blocked because serialising to disk/network defeats the whole point of the wrapper.
- copy / deepcopy return a fresh Secret (same value) so object-graph copies remain safe.
Usage
token = Secret(os.environ["API_TOKEN"]) headers = {"Authorization": f"Bearer {token.get()}"} # explicit .get() logging.info("request headers: %s", headers) # *** in logs
__bool__()
¶
Allow truthiness checks without exposing the secret value.
Enables guard clauses like if ctx.token: without leaking the
value. An empty string is falsy; any non-empty string is truthy.
Returns:
| Type | Description |
|---|---|
bool
|
True if the wrapped value is non-empty, False otherwise. |
__copy__()
¶
Return a new Secret wrapping the same value.
copy.copy() calls copy when available. We return a new Secret so the copy is still opaque and slots-protected.
Returns:
| Type | Description |
|---|---|
Secret
|
A new Secret instance with the same underlying value. |
__deepcopy__(memo)
¶
Return a new Secret wrapping the same value (deep copy is shallow here).
copy.deepcopy() follows deepcopy. Strings are immutable, so a shallow copy of the value is always correct and avoids any attempt to recurse into the slot.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
memo
|
dict[int, Any]
|
The deepcopy memo dictionary (unused, but required by protocol). |
required |
Returns:
| Type | Description |
|---|---|
Secret
|
A new Secret instance with the same underlying value. |
__format__(format_spec)
¶
Return a redacted placeholder for format(secret, spec) and f"{secret}".
This covers calls such as format(secret, spec), f"{secret}",
and f"{secret:spec}". The !r f-string conversion is handled
by __repr__ before formatting, so f"{secret!r}" is protected
by __repr__, not by this method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
format_spec
|
str
|
The format specification string (ignored). |
required |
Returns:
| Type | Description |
|---|---|
str
|
The literal string |
__init__(value)
¶
Wrap a sensitive string value in an opaque, redaction-safe container.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
str
|
The raw secret string (e.g., an API token or password). |
required |
__reduce__()
¶
Block pickling to prevent accidental serialisation of secrets.
pickle calls reduce to determine how to serialise an object. Raising TypeError here stops pickle.dumps() before it writes anything to disk or the network.
Raises:
| Type | Description |
|---|---|
TypeError
|
Always -- Secret instances cannot be pickled. |
__repr__()
¶
Return a safe repr that omits the secret value.
Called by repr(), debuggers, pytest output, and dataclass repr. Includes the type name so developers know what they're looking at, but never includes the value itself.
Returns:
| Type | Description |
|---|---|
str
|
The literal string |
__str__()
¶
Return a redacted placeholder instead of the secret value.
Called by str(), format(), f-strings, and %-style formatting.
Returns:
| Type | Description |
|---|---|
str
|
The literal string |
get()
¶
Return the actual secret value.
Calling .get() is a deliberate act: reviewers know that any code path reaching this line is handling sensitive material.
Returns:
| Type | Description |
|---|---|
str
|
The unwrapped secret string. |
add_global_option(cli, *param_decls, **option_kwargs)
¶
Install a Click option at root + every group + every subcommand of cli.
The option is accepted at ANY level of the CLI hierarchy. The resolved
value is merged into the root Click context's meta dict under the
option's Python-identifier name (e.g., --foo-bar -> meta['foo_bar']).
Read it from command callbacks via::
# For add_global_option(cli, "--json", is_flag=True) the meta key
# is "json" (Click's standard param-decl-to-name derivation). Use
# whatever key your flag derives to -- ``--my-flag`` -> ``"my_flag"``,
# ``--api-url`` -> ``"api_url"``, etc.
root_meta = click.get_current_context().find_root().meta
is_json = root_meta["json"]
Merge semantics
- Plain flags (
is_flag=Truewith a single--foodeclaration) use OR across levels: a truthy value at ANY level wins, someta[name]isTrueif the user passed the flag at root OR group OR subcommand (or any combination). OR is the only sensible rule for plain flags: the user has no way to say "off" at an inner level, so treating truthy-anywhere as wins is what matches intuition. - Slash-flags (
is_flag=Truewith a--foo/--no-foodeclaration) are an exception to the OR rule. Because the user CAN say "off" at an inner level via--no-foo, we switch to innermost-wins so an inner--no-foocan override an outer--foo(and vice versa). If we OR'd these, False would never win and the--no-fooform would be useless at inner levels. - Value options (strings, ints, enums, ...) use innermost-wins:
the deepest level that explicitly supplied the option provides
the final value. "Explicit" here means any Click
ParameterSourceother thanDEFAULT-- command line is the common case, but environment variables anddefault_map-sourced values also count as explicit and can override outer levels. Levels that parsed only the Click default do NOT overwrite an already-set value. - Not passed anywhere:
meta[name]is the Click-resolved default -- typicallyFalsefor flags andNonefor value options, but callers can change either withdefault=...inoption_kwargs. If the caller declared a flag withdefault=True,meta[name]isTruewhen the user didn't pass the flag; the OR-merge semantics still apply on top of whatever default Click resolves at each level.
Snapshot behaviour
Registration is a call-time snapshot of cli's command tree.
Commands attached to cli AFTER add_global_option returns do
NOT retroactively receive the option. This is deliberate: retroactive
registration would require monkey-patching Group.add_command and
introduces lifecycle surprises. The correct workflow is to call
add_global_option ONCE, AFTER all commands (including those from
discover_commands / entry points) are attached. Re-invoking it
later with the same flag raises ValueError because the
conflict-detection guard treats the existing registration as a
collision -- this is intentional: silent idempotent reinstalls
would hide real "declared the same flag twice by mistake" bugs.
Lazy entry-point plugins
The install-time flag-string conflict check is best effort for
entry-point plugins wrapped in LazyEntryPointCommand. Those
proxies don't expose the plugin's own options until the plugin
module is actually loaded (that's the whole point of lazy
discovery), so add_global_option can't see an internal
--json declared by the plugin at install time. If the global
flag AND the plugin-internal option end up declaring the same
flag string, LazyEntryPointCommand.invoke catches the
collision at invocation time and raises click.UsageError
with both sides named. Plugin authors who hit that error should
rename their internal option OR the CLI author should skip the
global install for that specific flag.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
cli
|
Group
|
The root Click group to install the option on. Walked recursively to discover nested groups and leaf commands. |
required |
*param_decls
|
str
|
Click parameter declarations -- the same strings you'd
pass to |
()
|
**option_kwargs
|
Any
|
Keyword arguments forwarded to |
{}
|
Raises:
| Type | Description |
|---|---|
TypeError
|
If |
TypeError
|
If |
ValueError
|
If Click cannot derive a Python name from
|
ValueError
|
If any command or group in the tree already has an
option with the same Python name or flag string -- catches
both "called |
Examples:
Flag with OR semantics, read anywhere in your code::
cli = clickwork.create_cli(name="myapp", commands_dir=...)
clickwork.add_global_option(cli, "--json", is_flag=True,
help="Output as JSON.")
# All three invocations leave ctx.find_root().meta['json'] == True:
# myapp --json sub-cmd
# myapp sub-cmd --json
# myapp group --json sub-cmd
Value option with innermost-wins. Note this example uses
--region rather than --env: create_cli already
installs --env at the root (alongside --verbose,
--quiet, --dry-run, --yes), so calling
add_global_option(cli, "--env", ...) against a
create_cli root would raise ValueError for a
flag-string collision. Pick a name that is not one of the
clickwork-reserved built-ins::
clickwork.add_global_option(cli, "--region", default=None,
help="Target region.")
# myapp --region=us-east sub-cmd --region=eu-west
# => ctx.find_root().meta['region'] == 'eu-west' (inner wins)
# myapp --region=us-east sub-cmd
# => ctx.find_root().meta['region'] == 'us-east' (outer alone)
# myapp sub-cmd
# => ctx.find_root().meta['region'] is None (Click default)
create_cli(name, commands_dir=None, discovery_mode='auto', config_schema=None, repo_config_path=None, *, description=None, enable_parent_package_imports=False, strict=False, version=None, package_name=None)
¶
Create a Click CLI group with global flags and plugin discovery.
This is the main entry point for building a clickwork CLI. It returns a Click group that can be invoked directly or used as a console_scripts entry point.
The group has these global flags available to every subcommand
--verbose / -v (count, repeatable -- -v is INFO, -vv is DEBUG) --quiet / -q (flag -- suppress all non-error output) --dry-run (flag -- preview without executing) --env (string -- select config environment) --yes / -y (flag -- skip confirmation prompts)
If version or package_name is provided, the CLI also gains
-V / --version at the root level (wired via
:func:click.version_option). When neither kwarg is set, no
version flag is installed -- existing callers see no change.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
CLI name (e.g., "orbit-admin"). Used for config paths and logging. |
required |
commands_dir
|
Path | None
|
Path to the commands directory for dev-mode discovery. |
None
|
discovery_mode
|
str
|
"dev", "installed", or "auto". |
'auto'
|
config_schema
|
dict[str, Any] | None
|
Optional config schema dict for validation. |
None
|
repo_config_path
|
Path | None
|
Optional path to repo-level config file. |
None
|
description
|
str | None
|
Short help summary shown at the top of |
None
|
enable_parent_package_imports
|
bool
|
When True (and |
False
|
strict
|
bool
|
When True, any command-discovery failure raises
Scope note: in Use this flag for production CLIs and release validation where shipping a binary with a missing command is a bug. Defaults to False to preserve the forgiving dev-mode behaviour so upgraders see no change unless they opt in. Keyword-only to keep the positional signature stable. See issue #42 for the full rationale. |
False
|
version
|
str | None
|
Explicit version string (e.g. |
None
|
package_name
|
str | None
|
Installed distribution name to auto-resolve the
version from via :func: |
None
|
Returns:
| Type | Description |
|---|---|
Group
|
A configured Click group with all discovered commands registered. |
Raises:
| Type | Description |
|---|---|
ClickworkDiscoveryError
|
If |
ValueError
|
If |
Example
Explicit version string::
cli = create_cli(name="orbit-admin", version="1.2.3")
Auto-resolve from the installing package's metadata::
cli = create_cli(name="orbit-admin", package_name="orbit-admin")
delete(url, *, body=None, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP DELETE. See :func:get / :func:post for shared kwargs.
DELETE with a body is unusual but legal (some APIs use it for bulk delete payloads); we accept it symmetrically with PUT/POST rather than forcing a different signature.
get(url, *, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP GET and return the parsed JSON or raw bytes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
Absolute URL ( |
required |
allowed_hosts
|
list[str] | None
|
Optional per-call URL allowlist. |
None
|
bearer_token
|
str | Secret | None
|
Optional bearer token ( |
None
|
basic_auth
|
tuple[str, str | Secret] | None
|
Optional |
None
|
headers
|
dict[str, str] | None
|
Optional extra request headers. An explicit
|
None
|
parse_json
|
bool
|
When True (default), auto-parse responses whose
Content-Type matches |
True
|
timeout
|
float
|
Socket-level timeout in seconds. |
30.0
|
Returns:
| Type | Description |
|---|---|
JSONValue | bytes
|
The parsed JSON value ( |
JSONValue | bytes
|
and |
Raises:
| Type | Description |
|---|---|
ValueError
|
On allowlist mismatch (before any network traffic). |
HttpError
|
On any non-2xx HTTP response. |
URLError
|
On transport failures (timeout, DNS, etc.). |
load_config(project_name, repo_config_path=None, user_config_path=None, env=None, schema=None)
¶
Load and merge config from all sources.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
project_name
|
str
|
CLI project name (e.g., "orbit-admin"). Used for env var prefix and default config file paths. |
required |
repo_config_path
|
Path | None
|
Path to repo-level config (.orbit-admin.toml). If None, looks for .{project_name}.toml in cwd. |
None
|
user_config_path
|
Path | None
|
Path to user-level config. If None, uses ~/.config/{project_name}/config.toml. |
None
|
env
|
str | None
|
Selected environment (e.g., "staging"). Falls back to the {PROJECT_NAME}_ENV env var when --env is omitted (i.e., env is None). |
None
|
schema
|
dict[str, Any] | None
|
Optional config schema dict for validation. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Merged config dict with all keys resolved. |
Raises:
| Type | Description |
|---|---|
ConfigError
|
If schema validation fails (missing required key, secret in repo config, type mismatch, unsafe permissions). |
normalize_prefix(name)
¶
Convert a project/CLI name to a shell-safe env-var prefix.
Hyphens become underscores and the result is uppercased so the prefix
conforms to POSIX env-var naming rules (e.g., orbit-admin ->
ORBIT_ADMIN). Used by both the CLI harness (for {PREFIX}_ENV
resolution) and the config loader (for auto-prefixed env vars).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
A project or CLI name, possibly containing hyphens. |
required |
Returns:
| Type | Description |
|---|---|
str
|
An uppercase, underscore-delimited prefix string. |
pass_cli_context(f)
¶
Decorator that injects a CliContext into a Click command function.
Safer than @click.pass_obj because it traverses the full Click
context chain with find_object(CliContext) (works in deeply nested
groups) and raises a descriptive UsageError instead of letting commands
crash with AttributeError: 'NoneType' has no attribute 'dry_run'
when the CLI was not created through create_cli().
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
f
|
Callable[..., Any]
|
The Click command function to wrap. Its first positional argument must be typed as CliContext. |
required |
Returns:
| Type | Description |
|---|---|
Callable[..., Any]
|
A wrapped function compatible with Click's decorator stack. |
Usage
@click.command() @pass_cli_context def deploy(ctx: CliContext) -> None: ctx.run(["kubectl", "apply", "-f", "manifests/"])
platform_dispatch(*, linux=None, windows=None, macos=None, linux_error=None, windows_error=None, macos_error=None)
¶
Decorate a command so it dispatches to a per-OS implementation at call time.
The decorated function's body is never executed -- it exists purely to
carry the Click decorator stack (@click.command, @click.argument,
@click.option, @pass_cli_context, etc.) and define the signature
that each platform impl must satisfy. At call time, the decorator detects
the current platform via sys.platform (using is_linux/
is_windows/is_macos) and forwards the caller's args/kwargs to the
matching impl.
The three *_error kwargs are part of the public API with no macOS
carve-out -- any platform can opt out of support by passing
<platform>=None plus an optional <platform>_error="..." message.
When a custom message is not provided, the default is
"<platform> not supported". The error is raised as
click.UsageError so Click prints it cleanly and exits with code 2,
matching clickwork's "user error, not framework bug" policy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
linux
|
Callable[..., Any] | None
|
Implementation to run on Linux ( |
None
|
windows
|
Callable[..., Any] | None
|
Implementation to run on Windows ( |
None
|
macos
|
Callable[..., Any] | None
|
Implementation to run on macOS ( |
None
|
linux_error
|
str | None
|
Custom UsageError message when invoked on linux with
|
None
|
windows_error
|
str | None
|
Custom UsageError message when invoked on windows with
|
None
|
macos_error
|
str | None
|
Custom UsageError message when invoked on macos with
|
None
|
Returns:
| Type | Description |
|---|---|
Callable[[Callable[..., Any]], Callable[..., Any]]
|
A decorator that replaces the wrapped function with a dispatcher. |
Callable[[Callable[..., Any]], Callable[..., Any]]
|
The dispatcher preserves the wrapped function's metadata via |
Callable[[Callable[..., Any]], Callable[..., Any]]
|
|
Example::
# Decorator ORDER matters. platform_dispatch never calls the
# wrapped function's body -- it only uses it to carry the Click
# command metadata (name, help, args/options). That means any
# decorator applied ABOVE platform_dispatch (outer, applied
# later) sees platform_dispatch's dispatcher as the callable,
# and any decorator applied BELOW platform_dispatch (inner,
# applied earlier) never runs because platform_dispatch discards
# the body.
#
# Practical rule: put platform_dispatch at the *bottom* of the
# decorator stack (closest to ``def``) so ``@pass_cli_context``
# and any other callback-wrapping decorators have already added
# their injection logic to the enclosing stack. Click's
# ``@click.command()`` / ``@click.argument()`` go above, as usual.
@click.command()
@click.argument("name")
@pass_cli_context
@clickwork.platform_dispatch(
linux=my_lib.linux.up,
windows=my_lib.windows.up,
macos=my_lib.macos.up,
macos_error="macOS not supported yet",
)
def runner_up(ctx, name): ...
post(url, *, body=None, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP POST. See :func:get for shared kwargs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
body
|
JSONValue | bytes | None
|
The request body.
|
None
|
put(url, *, body=None, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP PUT. See :func:get / :func:post for shared kwargs.
clickwork.http¶
Stateless HTTP client built on the Python standard library.
Design tenets¶
stdlib only. This module uses urllib.request + urllib.parse +
json + base64 -- no requests, no httpx, no third-party HTTP
client. Clickwork is intended to be trivially embeddable in any Python 3.11+
project without pulling in an SSL-pinned dependency tree. If a caller needs
requests-style conveniences (sessions, connection pooling, complex
retry policies), they can add that dependency themselves at the project
level; we refuse to make the choice for them.
Allowlist opt-in. allowed_hosts defaults to None (disabled). When
populated, the URL's .hostname is compared case-insensitively against
each entry and a ValueError is raised before any network activity
happens on a mismatch. We raise ValueError (not HttpError) for these
pre-flight rejections because no HTTP status exists for a request that
never left the process -- HttpError is reserved for actual server
non-2xx responses. Operators who want the safety net opt in per-call by
passing a populated list; the rest of the world gets urllib's default
behaviour.
Auth precedence. If the caller's headers dict already contains
Authorization, that wins over everything. Otherwise, bearer_token
produces Authorization: Bearer <token> and basic_auth produces
Authorization: Basic <base64(user:pw)>. Both accept either str or
:class:~clickwork._types.Secret; unwrapping happens only at the moment
the header value is built, and the unwrapped value is never logged.
JSON auto-parse. Responses whose Content-Type media type is
application/json (case-insensitive; parameters like
; charset=utf-8 are allowed after the media type)
are json.loads-decoded when parse_json=True (the default). Any other
Content-Type -- or parse_json=False -- yields raw bytes. The return
type is therefore the union JSONValue | bytes; narrow at the call site
with an isinstance check or a typing.cast.
Redaction policy. Each request emits exactly one log line at INFO level::
GET https://api.example.com/v1/foo [auth: <redacted>]
If no auth was attached, the [auth: ...] suffix is omitted entirely.
Token and password values NEVER appear in log output. Secret.get()
is called at most once per request and only inside header construction.
Error model. Non-2xx responses arrive via urllib.error.HTTPError;
we catch them and re-raise as :class:HttpError with all four attributes
populated: status_code, headers, url, and response_body
(parsed as JSON when the error response's Content-Type matched, else
bytes). Transport-level errors (timeouts, DNS failures, connection
refused) are NOT caught -- they propagate as the underlying
urllib.error.URLError subclass so callers can distinguish "the server
said no" (HttpError) from "we never reached a server" (URLError). Catch
pattern::
try:
data = http.get(url, allowed_hosts=[...])
except HttpError as e:
if e.status_code == 404:
... # handle 404 specifically
raise
except URLError:
... # network/transport issue -- retry, fail over, etc.
HttpError
¶
Bases: Exception
Raised when the server returns a non-2xx response.
Mirrors the :class:~clickwork._types.CliProcessError pattern: raw
fields are exposed as structured attributes so callers can branch on
status_code without parsing a string message.
Attributes:
| Name | Type | Description |
|---|---|---|
status_code |
int
|
HTTP status code reported by the server (e.g. 404, 500). |
response_body |
JSONValue | bytes
|
Parsed JSON if the original request was issued with
|
headers |
dict[str, str]
|
Response headers as a plain |
url |
str
|
The URL that produced this error -- useful for logs when a single caller issues multiple requests in a loop. |
Usage::
try:
http.get(url, allowed_hosts=[...])
except HttpError as e:
if e.status_code == 404:
return None # "not found" is expected here
raise # anything else is a real error
__init__(status_code, response_body, headers, url, message)
¶
Build an HttpError with the full context needed for triage.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
status_code
|
int
|
HTTP status from the server. |
required |
response_body
|
JSONValue | bytes
|
Parsed JSON body (when the caller opted into
|
required |
headers
|
dict[str, str]
|
Response headers as a plain string-keyed dict. |
required |
url
|
str
|
The URL that produced this error. |
required |
message
|
str
|
A pre-composed human-readable message for |
required |
delete(url, *, body=None, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP DELETE. See :func:get / :func:post for shared kwargs.
DELETE with a body is unusual but legal (some APIs use it for bulk delete payloads); we accept it symmetrically with PUT/POST rather than forcing a different signature.
get(url, *, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP GET and return the parsed JSON or raw bytes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
Absolute URL ( |
required |
allowed_hosts
|
list[str] | None
|
Optional per-call URL allowlist. |
None
|
bearer_token
|
str | Secret | None
|
Optional bearer token ( |
None
|
basic_auth
|
tuple[str, str | Secret] | None
|
Optional |
None
|
headers
|
dict[str, str] | None
|
Optional extra request headers. An explicit
|
None
|
parse_json
|
bool
|
When True (default), auto-parse responses whose
Content-Type matches |
True
|
timeout
|
float
|
Socket-level timeout in seconds. |
30.0
|
Returns:
| Type | Description |
|---|---|
JSONValue | bytes
|
The parsed JSON value ( |
JSONValue | bytes
|
and |
Raises:
| Type | Description |
|---|---|
ValueError
|
On allowlist mismatch (before any network traffic). |
HttpError
|
On any non-2xx HTTP response. |
URLError
|
On transport failures (timeout, DNS, etc.). |
post(url, *, body=None, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP POST. See :func:get for shared kwargs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
body
|
JSONValue | bytes | None
|
The request body.
|
None
|
put(url, *, body=None, allowed_hosts=None, bearer_token=None, basic_auth=None, headers=None, parse_json=True, timeout=30.0)
¶
Issue an HTTP PUT. See :func:get / :func:post for shared kwargs.
clickwork.platform¶
Platform detection, dispatch, and repository root finding.
Platform helpers (is_linux, is_macos, is_windows) are thin wrappers around
sys.platform. They exist so command code reads clearly: if is_macos():
instead of if sys.platform == "darwin":.
platform_dispatch (decorator) and dispatch() (functional helper) route a
single command to the right per-OS implementation at call time. The two
forms share _select_impl so the detection + error-message logic is
single-sourced and can't drift.
find_repo_root() walks up from a starting directory looking for .git as
either a directory (normal repo) or a file (worktree/submodule). Falls back
to git rev-parse --show-toplevel if the walk fails.
dispatch(ctx, *, linux=None, windows=None, macos=None, linux_error=None, windows_error=None, macos_error=None, **kwargs)
¶
Functional escape hatch for platform-dispatching from inside a command.
Use this form when you need to run pre-dispatch logic (loading config,
validating args, printing a banner) before branching on platform. The
decorator form (@platform_dispatch) is the primary surface; reach
for dispatch only when the decorator's "body-is-never-run" semantics
get in the way.
The selected impl is called as impl(ctx, **kwargs) -- ctx is
always forwarded as the first positional argument, matching the shape of
@pass_cli_context command callbacks. Any extra keyword arguments
passed to dispatch are forwarded to the impl alongside ctx.
The three *_error kwargs behave exactly like the decorator form:
when the matching impl kwarg is None, click.UsageError is raised
with the custom message if provided or f"{platform} not supported"
by default.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ctx
|
Any
|
The command's context object (typically a |
required |
linux
|
Callable[..., Any] | None
|
Impl to call when |
None
|
windows
|
Callable[..., Any] | None
|
Impl to call when |
None
|
macos
|
Callable[..., Any] | None
|
Impl to call when |
None
|
linux_error
|
str | None
|
Custom UsageError message for |
None
|
windows_error
|
str | None
|
Custom UsageError message for |
None
|
macos_error
|
str | None
|
Custom UsageError message for |
None
|
**kwargs
|
Any
|
Forwarded verbatim to the selected impl as keyword arguments. |
{}
|
Returns:
| Type | Description |
|---|---|
Any
|
Whatever the selected impl returns. |
Raises:
| Type | Description |
|---|---|
UsageError
|
If the current platform is unknown, or if the impl kwarg for the current platform is None. |
Example::
def runner_up(ctx, name: str) -> None:
ctx.logger.info("starting runner %s", name)
clickwork.platform.dispatch(
ctx,
linux=my_lib.linux.up,
windows=my_lib.windows.up,
macos=my_lib.macos.up,
macos_error="macOS not supported yet",
name=name,
)
find_repo_root(start=None)
¶
Walk up the directory tree to locate the repository root.
Searches for a .git entry (directory or file) starting from start
and traversing toward the filesystem root. Handles:
- Normal repos:
.gitis a directory. - Worktrees and submodules:
.gitis a file containinggitdir: ....
Falls back to git rev-parse --show-toplevel if the directory walk
does not find .git, which covers edge cases like bare repos.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
start
|
Path | None
|
Directory to begin the search from. Defaults to the current working directory when None. |
None
|
Returns:
| Type | Description |
|---|---|
Path | None
|
The absolute Path to the repository root, or None if not found. |
is_linux()
¶
Return True if the current platform is Linux.
Returns:
| Type | Description |
|---|---|
bool
|
True when sys.platform is |
is_macos()
¶
Return True if the current platform is macOS.
Returns:
| Type | Description |
|---|---|
bool
|
True when sys.platform is |
is_windows()
¶
Return True if the current platform is Windows.
Returns:
| Type | Description |
|---|---|
bool
|
True when sys.platform is |
platform_dispatch(*, linux=None, windows=None, macos=None, linux_error=None, windows_error=None, macos_error=None)
¶
Decorate a command so it dispatches to a per-OS implementation at call time.
The decorated function's body is never executed -- it exists purely to
carry the Click decorator stack (@click.command, @click.argument,
@click.option, @pass_cli_context, etc.) and define the signature
that each platform impl must satisfy. At call time, the decorator detects
the current platform via sys.platform (using is_linux/
is_windows/is_macos) and forwards the caller's args/kwargs to the
matching impl.
The three *_error kwargs are part of the public API with no macOS
carve-out -- any platform can opt out of support by passing
<platform>=None plus an optional <platform>_error="..." message.
When a custom message is not provided, the default is
"<platform> not supported". The error is raised as
click.UsageError so Click prints it cleanly and exits with code 2,
matching clickwork's "user error, not framework bug" policy.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
linux
|
Callable[..., Any] | None
|
Implementation to run on Linux ( |
None
|
windows
|
Callable[..., Any] | None
|
Implementation to run on Windows ( |
None
|
macos
|
Callable[..., Any] | None
|
Implementation to run on macOS ( |
None
|
linux_error
|
str | None
|
Custom UsageError message when invoked on linux with
|
None
|
windows_error
|
str | None
|
Custom UsageError message when invoked on windows with
|
None
|
macos_error
|
str | None
|
Custom UsageError message when invoked on macos with
|
None
|
Returns:
| Type | Description |
|---|---|
Callable[[Callable[..., Any]], Callable[..., Any]]
|
A decorator that replaces the wrapped function with a dispatcher. |
Callable[[Callable[..., Any]], Callable[..., Any]]
|
The dispatcher preserves the wrapped function's metadata via |
Callable[[Callable[..., Any]], Callable[..., Any]]
|
|
Example::
# Decorator ORDER matters. platform_dispatch never calls the
# wrapped function's body -- it only uses it to carry the Click
# command metadata (name, help, args/options). That means any
# decorator applied ABOVE platform_dispatch (outer, applied
# later) sees platform_dispatch's dispatcher as the callable,
# and any decorator applied BELOW platform_dispatch (inner,
# applied earlier) never runs because platform_dispatch discards
# the body.
#
# Practical rule: put platform_dispatch at the *bottom* of the
# decorator stack (closest to ``def``) so ``@pass_cli_context``
# and any other callback-wrapping decorators have already added
# their injection logic to the enclosing stack. Click's
# ``@click.command()`` / ``@click.argument()`` go above, as usual.
@click.command()
@click.argument("name")
@pass_cli_context
@clickwork.platform_dispatch(
linux=my_lib.linux.up,
windows=my_lib.windows.up,
macos=my_lib.macos.up,
macos_error="macOS not supported yet",
)
def runner_up(ctx, name): ...
clickwork.testing¶
Test helpers for clickwork-built CLIs.
This module provides two thin wrappers over Click's own testing toolkit so
plugin test suites can stop re-typing the same 4-line setup. It ships on
purpose with a minimal surface -- run_cli and make_test_cli only --
because most test authors benefit from keeping the rest of Click's testing
API visible. If you want to stub out subprocess calls, build your own mocks;
we deliberately do not ship mock_run / mock_capture context managers.
Why these helpers exist¶
Before this module existed, every plugin test that exercised a full CLI looked like::
from click.testing import CliRunner
from clickwork import create_cli
def test_greet(tmp_path):
(tmp_path / "greet.py").write_text(...)
cli = create_cli(name="test-cli", commands_dir=tmp_path)
runner = CliRunner()
result = runner.invoke(cli, ["greet"], catch_exceptions=False)
assert result.exit_code == 0
Two pieces of boilerplate appeared in every file: constructing the CLI with
a throwaway name, and remembering to pass catch_exceptions=False so real
tracebacks surface in test output. This module collapses both.
Canonical usage¶
::
from clickwork.testing import make_test_cli, run_cli
def test_greet_says_hello(tmp_path):
(tmp_path / "greet.py").write_text(
"import click\n"
"@click.command()\n"
"def greet():\n"
" click.echo('hello')\n"
"cli = greet\n"
)
cli = make_test_cli(commands_dir=tmp_path)
result = run_cli(cli, ["greet"])
assert result.exit_code == 0
assert "hello" in result.stdout
CliRunner output attributes -- pick the right one¶
Click's :class:click.testing.Result exposes three stream attributes that
look similar but differ in subtle ways that matter for assertion tests:
result.output-- stdout AND stderr interleaved in the order the command produced them. Convenient for "did the command say X at all" smoke checks. Misleading for "did the error go to stderr" tests, because the answer is always "yes, and also it's inoutput".result.stdout-- stdout only. Assert on this when the contract you care about is specifically "this goes to the normal output channel".result.stderr-- stderr only. Assert on this when the contract is "this is an error / diagnostic / progress line on a side channel".
A test that says "the error message was printed to stderr" should assert
on result.stderr, not result.output. See the GUIDE.md "Testing
commands" section for a worked example.
Historical note: Click 8.2 removed the mix_stderr kwarg that
CliRunner.__init__ used to accept. Post-removal, all three stream
attributes on Result are populated separately (output is the
interleaved form; stdout and stderr are kept independent).
clickwork declares click>=8.2 precisely so this guidance always
applies -- snippets in older tutorials that use
CliRunner(mix_stderr=False) will raise TypeError against the
supported Click range, and on 8.1 and earlier result.stderr would
have raised ValueError: stderr not separately captured under the
default CliRunner() configuration (where streams were mixed unless
mix_stderr=False was passed explicitly). Flooring at 8.2 gets us
out of documenting that conditional behaviour.
make_test_cli(*, commands_dir=None, **create_cli_kwargs)
¶
Build a clickwork CLI with sensible test-suite defaults.
Thin convenience wrapper over :func:clickwork.create_cli. Fills in a
default name ("test-cli") so tests that don't care about the
CLI name don't have to repeat that argument, and forwards every other
kwarg through unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
commands_dir
|
Path | None
|
Directory containing command |
None
|
**create_cli_kwargs
|
Any
|
Forwarded verbatim to |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
A |
Group
|
class: |
Example::
cli = make_test_cli(commands_dir=tmp_path, description="deploy helpers")
result = run_cli(cli, ["--help"])
assert "deploy helpers" in result.stdout
run_cli(cli, args=None, **kwargs)
¶
Invoke a Click CLI under CliRunner with test-friendly defaults.
Equivalent to click.testing.CliRunner().invoke(cli, args, **kwargs)
with one change: catch_exceptions defaults to False so bugs in
the command surface as real tracebacks in pytest output instead of
being swallowed into result.exception. Pass catch_exceptions=True
explicitly if you want to assert on the caught exception.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
cli
|
Command
|
The Click command or group to invoke. Accepts any
|
required |
args
|
str | Sequence[str] | None
|
The command-line arguments to pass, as you would write them
after the CLI name. The preferred form is a list/tuple of
already-tokenised strings ( |
None
|
**kwargs
|
Any
|
Forwarded verbatim to |
{}
|
Returns:
| Type | Description |
|---|---|
Result
|
Click's native :class: |
Result
|
not wrap this -- plugin authors already know its shape. |
Example::
result = run_cli(cli, ["deploy", "--dry-run"])
assert result.exit_code == 0
assert "would deploy" in result.stdout
clickwork.config¶
Layered TOML configuration with environment support.
Config is loaded from multiple sources with cascading precedence (highest wins):
1. Environment variables (explicit mapping or auto-prefixed)
2. Env-specific section ([env.staging]) in repo config
3. [default] section in repo config (.{project-name}.toml)
4. User-level config (~/.config/{project-name}/config.toml)
The env-specific section overrides [default] but doesn't replace it -- keys not specified in the env section fall through to [default].
Schema validation (optional) ensures required keys exist, types match, and secrets don't leak into repo config. User config with loose permissions is refused (not just warned) to prevent secret leakage.
ConfigError
¶
Bases: Exception
Raised when config validation fails.
This is a user-facing error -- the message should be actionable, telling them which key is missing/invalid and where to fix it.
load_config(project_name, repo_config_path=None, user_config_path=None, env=None, schema=None)
¶
Load and merge config from all sources.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
project_name
|
str
|
CLI project name (e.g., "orbit-admin"). Used for env var prefix and default config file paths. |
required |
repo_config_path
|
Path | None
|
Path to repo-level config (.orbit-admin.toml). If None, looks for .{project_name}.toml in cwd. |
None
|
user_config_path
|
Path | None
|
Path to user-level config. If None, uses ~/.config/{project_name}/config.toml. |
None
|
env
|
str | None
|
Selected environment (e.g., "staging"). Falls back to the {PROJECT_NAME}_ENV env var when --env is omitted (i.e., env is None). |
None
|
schema
|
dict[str, Any] | None
|
Optional config schema dict for validation. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
Merged config dict with all keys resolved. |
Raises:
| Type | Description |
|---|---|
ConfigError
|
If schema validation fails (missing required key, secret in repo config, type mismatch, unsafe permissions). |
load_env_file(path)
¶
Parse a dotenv-style file into a plain dict of string key/value pairs.
This is a standalone helper -- it is not integrated into
load_config(). The TOML pipeline handles structured config; this
helper covers the separate case where a command needs to source
credentials or environment variables from a .env file and pass
them to a subprocess (ctx.run(env=...)) or inject them into
os.environ.
Supported syntax (deliberately tiny):
KEY=value # simple assignment
export KEY=value # optional shell-style 'export' prefix, stripped
KEY="value with ws" # double-quoted value (quotes stripped)
KEY='value with ws' # single-quoted value (quotes stripped)
# full-line comment # skipped
<blank line> # skipped
Explicitly NOT supported (by design -- do not add these):
* Variable substitution. ``K=$OTHER`` stores the literal string
``"$OTHER"``; nothing is resolved from ``os.environ`` or from
earlier entries in the file. If you need shell semantics, use
``sh -c 'set -a; source .env; env'`` and capture stdout.
* Backticks / command substitution (e.g. ``K=$(date)`` or the
backtick form).
* Heredocs and multi-line values.
* Inline trailing comments (``K=val # comment`` is parsed as
``K`` -> ``"val # comment"`` because the literal '#' is part of
the value, not a comment marker).
These omissions are intentional. A tiny grammar is a feature: it means
you can look at a .env file and know exactly what each line does
without reading the parser.
Security: the file must be owner-only (chmod 600) on POSIX
platforms. Because .env files typically hold secrets, any
group or other permission bit (read, write, OR execute) raises
ConfigError -- not just group/other readability. A group-writable
file is a tampering risk even when not group-readable, so the
rejection matches what chmod 600 actually enforces. On Windows
the check is skipped -- POSIX mode bits do not apply there, and
callers are expected to protect the file via NTFS ACLs instead.
Example usage::
from pathlib import Path
from clickwork.config import load_env_file
env = load_env_file(Path(".env"))
# env == {"API_TOKEN": "...", "REGION": "us-east-1"}
# Pass to a subprocess without mutating the parent environment.
# Note the list form: ctx.run (and the underlying subprocess
# helpers) reject string commands as a shell-injection guardrail,
# so every cmd must be an argv list.
ctx.run(["./deploy.sh"], env={**os.environ, **env})
# Or inject into the parent process:
os.environ.update(env)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Path
|
Path to the dotenv file to parse. Must exist (unlike user config, a missing .env is an error -- the caller asked for this file by name). |
required |
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
Dict mapping keys to their (possibly unquoted) string values. |
dict[str, str]
|
Insertion order matches the order each key first appears in |
dict[str, str]
|
the file. If the file contains the same key twice, the later |
dict[str, str]
|
value overwrites the earlier one but the dict slot stays at |
dict[str, str]
|
the first occurrence's position -- if caller cares about order, |
dict[str, str]
|
they should ensure no duplicate keys in the file. |
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
If |
ConfigError
|
If the file has unsafe permissions (POSIX only), or
if any line is malformed -- missing |