mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
4ee2b259c9
- Add build-context helper to copy Docker context safely (ignore runtime data, keep symlinks) - Update UI and core actions to use ~/.prole/build and shared copy helper - Add/adjust tests and scripts; introduce knoe ops helpers and update manifests Co-authored-by: Junie <junie@jetbrains.com>
96 lines
2.2 KiB
Python
96 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
from .context import KnoeContext
|
|
from .process import CommandFailed, check
|
|
|
|
|
|
def _env(ctx: KnoeContext, extra_env: Mapping[str, str] | None = None) -> dict[str, str]:
|
|
e = dict(os.environ)
|
|
e.update(ctx.env)
|
|
if extra_env:
|
|
e.update(dict(extra_env))
|
|
return e
|
|
|
|
|
|
def kubectl(
|
|
ctx: KnoeContext,
|
|
*args: str,
|
|
namespace: str | None = None,
|
|
input_text: str | None = None,
|
|
extra_env: Mapping[str, str] | None = None,
|
|
) -> str:
|
|
cmd: list[str] = ["kubectl", *args]
|
|
if namespace and ("-n" not in args and "--namespace" not in args):
|
|
cmd += ["-n", namespace]
|
|
proc = check(cmd, env=_env(ctx, extra_env), input_text=input_text)
|
|
return proc.stdout
|
|
|
|
|
|
def kubectl_json(ctx: KnoeContext, *args: str, namespace: str | None = None) -> Any:
|
|
out = kubectl(ctx, *args, "-o", "json", namespace=namespace)
|
|
return json.loads(out or "{}")
|
|
|
|
|
|
def namespace_exists(ctx: KnoeContext, namespace: str) -> bool:
|
|
try:
|
|
kubectl(ctx, "get", "namespace", namespace)
|
|
return True
|
|
except CommandFailed:
|
|
return False
|
|
|
|
|
|
def ensure_namespace(ctx: KnoeContext, namespace: str) -> None:
|
|
if namespace_exists(ctx, namespace):
|
|
return
|
|
kubectl(ctx, "create", "namespace", namespace)
|
|
|
|
|
|
def apply_yaml(ctx: KnoeContext, yaml_text: str, namespace: str | None = None) -> None:
|
|
kubectl(
|
|
ctx,
|
|
"apply",
|
|
"-f",
|
|
"-",
|
|
namespace=namespace,
|
|
input_text=yaml_text,
|
|
)
|
|
|
|
|
|
def get_resource_json(
|
|
ctx: KnoeContext,
|
|
resource: str,
|
|
name: str | None = None,
|
|
*,
|
|
namespace: str | None = None,
|
|
) -> Any:
|
|
args: list[str] = ["get", resource]
|
|
if name:
|
|
args.append(name)
|
|
return kubectl_json(ctx, *args, namespace=namespace)
|
|
|
|
|
|
def patch_resource(
|
|
ctx: KnoeContext,
|
|
resource: str,
|
|
name: str,
|
|
patch_json: str,
|
|
*,
|
|
namespace: str | None = None,
|
|
patch_type: str = "merge",
|
|
) -> None:
|
|
kubectl(
|
|
ctx,
|
|
"patch",
|
|
resource,
|
|
name,
|
|
"--type",
|
|
patch_type,
|
|
"-p",
|
|
patch_json,
|
|
namespace=namespace,
|
|
)
|