Coverage for .tox/coverage/lib/python3.13/site-packages/wuttjamaican/cli/runner.py: 100%
33 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-02 16:58 -0500
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-02 16:58 -0500
1# -*- coding: utf-8; -*-
2################################################################################
3#
4# WuttJamaican -- Base package for Wutta Framework
5# Copyright © 2023-2026 Lance Edgar
6#
7# This file is part of Wutta Framework.
8#
9# Wutta Framework is free software: you can redistribute it and/or modify it
10# under the terms of the GNU General Public License as published by the Free
11# Software Foundation, either version 3 of the License, or (at your option) any
12# later version.
13#
14# Wutta Framework is distributed in the hope that it will be useful, but
15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
17# more details.
18#
19# You should have received a copy of the GNU General Public License along with
20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.
21#
22################################################################################
23"""
24See also: :ref:`wutta-runner`
25"""
27import logging
28import subprocess
29import sys
31try:
32 from typing import Annotated
33except ImportError: # pragma: no cover
34 from typing_extensions import Annotated
36import humanize
37import typer
39from .base import wutta_typer
41log = logging.getLogger(__name__)
44@wutta_typer.command()
45def runner(
46 ctx: typer.Context,
47 subject: Annotated[
48 str,
49 typer.Option("--subject", "-S", help="Override value for email subject."),
50 ] = None,
51 skip_if_empty: Annotated[
52 bool,
53 typer.Option(
54 "--skip-if-empty",
55 help="Skip sending email if the command generates no output.",
56 ),
57 ] = False,
58 keep_exit_code: Annotated[
59 bool,
60 typer.Option(
61 "--keep-exit-code",
62 help="Exit with same return code as subprocess. "
63 "If not specified, exits with code 0 despite command failure.",
64 ),
65 ] = False,
66 command: Annotated[
67 str,
68 typer.Argument(help="Command to run, the result of which will be emailed."),
69 ] = ...,
70):
71 """
72 Run a command as subprocess, and email the result/output
73 """
74 config = ctx.parent.wutta_config
75 app = config.get_app()
77 # prep command
78 cmd = config.parse_list(command)
79 log.info("running command as subprocess: %s", cmd)
80 began = app.make_utc()
82 # run command
83 try:
84 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
85 retcode = 0
86 log.info("command completed okay")
87 except subprocess.CalledProcessError as error:
88 output = error.output
89 retcode = error.returncode
90 log.warning("command failed with code %s", retcode)
92 # capture timing
93 ended = app.make_utc()
94 runtime = ended - began
95 runtime_display = humanize.naturaldelta(runtime)
97 # convert output for readability
98 output = output.decode(errors="replace")
100 # maybe send email
101 if skip_if_empty and not output:
102 log.info("command had no output, so will skip sending email")
103 else:
104 app.send_email(
105 "runner_result",
106 {
107 "cmd": cmd,
108 "retcode": retcode,
109 "output": output,
110 "runtime": runtime,
111 "runtime_display": runtime_display,
112 },
113 subject=subject,
114 )
116 # exit with failure code if applicable
117 if retcode and keep_exit_code:
118 sys.exit(retcode)