"""Run command for the gyoza CLI."""
from __future__ import annotations
import ast
import importlib.util
import json
import os
from pathlib import Path
from types import ModuleType
from typing import Any
import typer
from gyoza.cli.errors import fail
_DEFAULT_INPUT_PATH = "/data/input.json"
_DEFAULT_OUTPUT_PATH = "/data/output.json"
_SEPARATOR = "━" * 50
def _load_module(path: Path) -> ModuleType:
"""Load and return a Python module from a file path."""
name = f"_gyoza_cli_{path.stem}_{abs(hash(path))}"
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
fail(f"unable to import module at {path}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _find_gyoza_op(module: ModuleType, func_name: str | None) -> Any:
"""
Return the gyoza op function from *module*.
Parameters
----------
module : ModuleType
Loaded Python module.
func_name : str or None
Explicit function name to look up. When ``None`` the first callable
with a ``.run`` attribute defined in the module is used.
Returns
-------
Any
The decorated function.
Raises
------
typer.Exit
On any lookup failure.
"""
if func_name:
func = getattr(module, func_name, None)
if func is None:
fail(f"function '{func_name}' not found in {module.__file__}")
return func
ops = [
obj
for obj in module.__dict__.values()
if callable(obj)
and hasattr(obj, "run")
and getattr(obj, "__module__", None) == module.__name__
]
if not ops:
fail(f"no @gyoza_op function found in {module.__file__}")
return ops[0]
def _parse_inline_input(raw: str) -> dict[str, Any]:
"""
Parse an inline string into a dict.
Parameters
----------
raw : str
JSON or Python-literal dict string.
Returns
-------
dict[str, Any]
Parsed input data.
Raises
------
typer.Exit
If the string cannot be parsed or is not a dict.
"""
try:
parsed = json.loads(raw)
except json.JSONDecodeError:
try:
parsed = ast.literal_eval(raw)
except (ValueError, SyntaxError) as exc:
fail(f"invalid --inline-input: {exc}")
if not isinstance(parsed, dict):
fail("--inline-input must be a JSON object (dictionary)")
return parsed
def _read_input_file(path: str) -> dict[str, Any]:
"""
Read a JSON input file into a dict.
Parameters
----------
path : str
Path to the JSON file.
Returns
-------
dict[str, Any]
Parsed input data.
Raises
------
typer.Exit
If the file cannot be read or parsed.
"""
try:
with Path(path).open() as fh:
data = json.load(fh)
except (OSError, json.JSONDecodeError) as exc:
fail(f"failed to read input file '{path}': {exc}")
if not isinstance(data, dict):
fail(f"input file '{path}' must contain a JSON object (dictionary)")
return data
def _write_output_file(path: str, data: dict[str, Any]) -> None:
"""
Write the output dict to a JSON file.
Parameters
----------
path : str
Destination file path.
data : dict[str, Any]
Output data to serialise.
Raises
------
typer.Exit
If the file cannot be written.
"""
try:
out = Path(path)
out.parent.mkdir(parents=True, exist_ok=True)
with out.open("w") as fh:
json.dump(data, fh, indent=2)
except OSError as exc:
fail(f"failed to write output file '{path}': {exc}")
[docs]
def run(
file: str | None = typer.Option(
None,
"--file",
"-f",
help="Python file that defines one @gyoza_op function.",
),
function: str | None = typer.Option(
None,
"--function",
"-F",
help=(
"Name of the @gyoza_op function to execute. "
"Defaults to the first one found in the file."
),
),
input: str | None = typer.Option(
None,
"--input",
"-i",
help=(
"JSON input file path. "
"Falls back to $GYOZA_INPUT_PATH. "
"Ignored when --inline-input is provided."
),
),
output: str | None = typer.Option(
None,
"--output",
"-o",
help=("JSON output file path. Falls back to $GYOZA_OUTPUT_PATH."),
),
inline_input: str | None = typer.Option(
None,
"--inline-input",
help=(
'Inline JSON dict string for inputs, e.g. \'{"a": 1, "b": 2}\'. '
"Mutually exclusive with --input."
),
),
) -> None:
"""
Execute a @gyoza_op function from a Python file.
Three input modes are supported:
1. ``--inline-input`` — parse input directly from the CLI string.
2. ``--input`` — read input from a JSON file at the given path.
3. default — read input from the path in $GYOZA_INPUT_PATH.
In all modes the output dict is written as JSON to ``--output`` or
$GYOZA_OUTPUT_PATH.
"""
if not file:
fail("--file is required")
if inline_input and input:
fail("--input and --inline-input are mutually exclusive")
code_path = Path(file).resolve()
if not code_path.exists():
fail(f"file not found: {file}")
resolved_output = output or os.getenv("GYOZA_OUTPUT_PATH", _DEFAULT_OUTPUT_PATH)
typer.echo(f"\n{_SEPARATOR}")
typer.echo(" ▶️ GYOZA RUN")
typer.echo(f"{_SEPARATOR}\n")
typer.echo(f" 📄 File: {code_path}")
typer.echo(f" 🎯 Function: {function or '(auto-detect)'}")
try:
module = _load_module(code_path)
func = _find_gyoza_op(module, function)
except typer.Exit:
raise
except Exception as exc: # noqa: BLE001
fail(f"failed to load module: {exc}")
if inline_input is not None:
mode = "inline"
input_dict = _parse_inline_input(inline_input)
typer.echo(f" 📝 Mode: {mode}")
typer.echo(f" 📥 Input: {inline_input}")
else:
resolved_input = input or os.getenv("GYOZA_INPUT_PATH", _DEFAULT_INPUT_PATH)
mode = "path" if input else "env"
typer.echo(f" 📝 Mode: {mode}")
typer.echo(f" 📥 Input: {resolved_input}")
input_dict = _read_input_file(resolved_input)
typer.echo(f" 📤 Output: {resolved_output}")
typer.echo(f"\n{_SEPARATOR}")
typer.echo(" 🚀 Executing...\n")
try:
output_dict = func.run(input_dict)
except Exception as exc: # noqa: BLE001
fail(f"op failed: {exc}")
_write_output_file(resolved_output, output_dict)
typer.echo(f"\n{_SEPARATOR}")
typer.echo(" ✅ Done.")
typer.echo(f"{_SEPARATOR}\n")
