mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 12:03:59 +00:00
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Callable, Mapping, Sequence
|
|
|
|
from .context import KnoeContext
|
|
from .process import check, run_streaming
|
|
|
|
|
|
def _env(ctx: KnoeContext, extra_env: Mapping[str, str] | None = None) -> dict[str, str]:
|
|
e = dict(os.environ)
|
|
e.update(ctx.env)
|
|
|
|
# Bridge common knobs for legacy scripts.
|
|
e.setdefault("KNOE_MODE", ctx.mode)
|
|
e.setdefault("PROLE_NAMESPACE", ctx.namespace)
|
|
e.setdefault("KNOE_SERVICE_NAMESPACE", ctx.service_namespace)
|
|
if ctx.cfg_path is not None:
|
|
e.setdefault("PROLE_CFG", str(ctx.cfg_path))
|
|
|
|
if extra_env:
|
|
e.update(dict(extra_env))
|
|
return e
|
|
|
|
|
|
def run_script(
|
|
ctx: KnoeContext,
|
|
script_name: str,
|
|
action: str,
|
|
*,
|
|
extra_args: Sequence[str] | None = None,
|
|
stdin_text: str | None = None,
|
|
on_line: Callable[[str], None] | None = None,
|
|
extra_env: Mapping[str, str] | None = None,
|
|
streaming: bool = True,
|
|
) -> int:
|
|
if not script_name.endswith(".sh"):
|
|
script_name = f"{script_name}.sh"
|
|
|
|
script_path = (ctx.project_root / "etc" / script_name).resolve()
|
|
cmd: list[str] = ["bash", str(script_path), action]
|
|
if extra_args:
|
|
cmd += list(extra_args)
|
|
|
|
env = _env(ctx, extra_env)
|
|
if streaming:
|
|
return run_streaming(cmd, cwd=ctx.project_root, env=env, stdin_text=stdin_text, on_line=on_line)
|
|
|
|
check(cmd, cwd=ctx.project_root, env=env, input_text=stdin_text)
|
|
return 0
|