Merge pull request 'cmd.py refactor part 5' (#724) from Qubasa-main into main
This commit is contained in:
@@ -1,9 +1,9 @@
|
|||||||
import argparse
|
import argparse
|
||||||
import json
|
import json
|
||||||
import subprocess
|
|
||||||
import sys
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
from ..cmd import run
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
from ..nix import nix_shell
|
from ..nix import nix_shell
|
||||||
from .secrets import encrypt_secret, sops_secrets_folder
|
from .secrets import encrypt_secret, sops_secrets_folder
|
||||||
@@ -23,10 +23,8 @@ def import_sops(args: argparse.Namespace) -> None:
|
|||||||
cmd += ["--input-type", args.input_type]
|
cmd += ["--input-type", args.input_type]
|
||||||
cmd += ["--output-type", "json", "--decrypt", args.sops_file]
|
cmd += ["--output-type", "json", "--decrypt", args.sops_file]
|
||||||
cmd = nix_shell(["nixpkgs#sops"], cmd)
|
cmd = nix_shell(["nixpkgs#sops"], cmd)
|
||||||
try:
|
|
||||||
res = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE)
|
res = run(cmd, error_msg=f"Could not import sops file {file}")
|
||||||
except subprocess.CalledProcessError as e:
|
|
||||||
raise ClanError(f"Could not import sops file {file}: {e}") from e
|
|
||||||
secrets = json.loads(res.stdout)
|
secrets = json.loads(res.stdout)
|
||||||
for k, v in secrets.items():
|
for k, v in secrets.items():
|
||||||
k = args.prefix + k
|
k = args.prefix + k
|
||||||
|
|||||||
@@ -2,12 +2,12 @@ import logging
|
|||||||
import os
|
import os
|
||||||
import shlex
|
import shlex
|
||||||
import shutil
|
import shutil
|
||||||
import subprocess
|
|
||||||
import sys
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from tempfile import TemporaryDirectory
|
from tempfile import TemporaryDirectory
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
from clan_cli.cmd import run
|
||||||
from clan_cli.nix import nix_shell
|
from clan_cli.nix import nix_shell
|
||||||
|
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
@@ -60,13 +60,9 @@ export facts={shlex.quote(str(facts_dir))}
|
|||||||
export secrets={shlex.quote(str(secrets_dir))}
|
export secrets={shlex.quote(str(secrets_dir))}
|
||||||
{generator}
|
{generator}
|
||||||
"""
|
"""
|
||||||
try:
|
cmd = nix_shell(["nixpkgs#bash"], ["bash", "-c", text])
|
||||||
cmd = nix_shell(["nixpkgs#bash"], ["bash", "-c", text])
|
run(cmd)
|
||||||
subprocess.run(cmd, check=True)
|
|
||||||
except subprocess.CalledProcessError:
|
|
||||||
msg = "failed to the following command:\n"
|
|
||||||
msg += text
|
|
||||||
raise ClanError(msg)
|
|
||||||
for name in secrets:
|
for name in secrets:
|
||||||
secret_file = secrets_dir / name
|
secret_file = secrets_dir / name
|
||||||
if not secret_file.is_file():
|
if not secret_file.is_file():
|
||||||
|
|||||||
Reference in New Issue
Block a user