use pathlib everywhere
This commit is contained in:
@@ -42,5 +42,5 @@ def remove_object(path: Path, name: str) -> list[Path]:
|
||||
msg = f"{name} not found in {path}"
|
||||
raise ClanError(msg) from e
|
||||
if not os.listdir(path):
|
||||
os.rmdir(path)
|
||||
path.rmdir()
|
||||
return paths_to_commit
|
||||
|
||||
@@ -122,7 +122,7 @@ def add_member(
|
||||
if not user_target.is_symlink():
|
||||
msg = f"Cannot add user {name}. {user_target} exists but is not a symlink"
|
||||
raise ClanError(msg)
|
||||
os.remove(user_target)
|
||||
user_target.unlink()
|
||||
user_target.symlink_to(os.path.relpath(source, user_target.parent))
|
||||
return update_group_keys(flake_dir, group_folder.parent.name)
|
||||
|
||||
@@ -133,16 +133,16 @@ def remove_member(flake_dir: Path, group_folder: Path, name: str) -> None:
|
||||
msg = f"{name} does not exist in group in {group_folder}: "
|
||||
msg += list_directory(group_folder)
|
||||
raise ClanError(msg)
|
||||
os.remove(target)
|
||||
target.unlink()
|
||||
|
||||
if len(os.listdir(group_folder)) > 0:
|
||||
update_group_keys(flake_dir, group_folder.parent.name)
|
||||
|
||||
if len(os.listdir(group_folder)) == 0:
|
||||
os.rmdir(group_folder)
|
||||
group_folder.rmdir()
|
||||
|
||||
if len(os.listdir(group_folder.parent)) == 0:
|
||||
os.rmdir(group_folder.parent)
|
||||
group_folder.parent.rmdir()
|
||||
|
||||
|
||||
def add_user(flake_dir: Path, group: str, name: str) -> None:
|
||||
|
||||
@@ -16,7 +16,7 @@ def extract_public_key(filepath: Path) -> str:
|
||||
Extracts the public key from a given text file.
|
||||
"""
|
||||
try:
|
||||
with open(filepath) as file:
|
||||
with filepath.open() as file:
|
||||
for line in file:
|
||||
# Check if the line contains the public key
|
||||
if line.startswith("# public key:"):
|
||||
|
||||
@@ -218,7 +218,7 @@ def allow_member(
|
||||
if not user_target.is_symlink():
|
||||
msg = f"Cannot add user '{name}' to {group_folder.parent.name} secret. {user_target} exists but is not a symlink"
|
||||
raise ClanError(msg)
|
||||
os.remove(user_target)
|
||||
user_target.unlink()
|
||||
|
||||
user_target.symlink_to(os.path.relpath(source, user_target.parent))
|
||||
changed = [user_target]
|
||||
@@ -244,13 +244,13 @@ def disallow_member(group_folder: Path, name: str) -> list[Path]:
|
||||
if len(keys) < 2:
|
||||
msg = f"Cannot remove {name} from {group_folder.parent.name}. No keys left. Use 'clan secrets remove {name}' to remove the secret."
|
||||
raise ClanError(msg)
|
||||
os.remove(target)
|
||||
target.unlink()
|
||||
|
||||
if len(os.listdir(group_folder)) == 0:
|
||||
os.rmdir(group_folder)
|
||||
group_folder.rmdir()
|
||||
|
||||
if len(os.listdir(group_folder.parent)) == 0:
|
||||
os.rmdir(group_folder.parent)
|
||||
group_folder.parent.rmdir()
|
||||
|
||||
return update_keys(
|
||||
target.parent.parent, sorted(collect_keys_for_path(group_folder.parent))
|
||||
@@ -337,7 +337,7 @@ def rename_command(args: argparse.Namespace) -> None:
|
||||
if new_path.exists():
|
||||
msg = f"Secret '{args.new_name}' already exists"
|
||||
raise ClanError(msg)
|
||||
os.rename(old_path, new_path)
|
||||
old_path.rename(new_path)
|
||||
commit_files(
|
||||
[old_path, new_path],
|
||||
flake_dir,
|
||||
|
||||
@@ -172,13 +172,11 @@ def encrypt_file(
|
||||
with NamedTemporaryFile(delete=False) as f:
|
||||
try:
|
||||
if isinstance(content, str):
|
||||
with open(f.name, "w") as fd:
|
||||
fd.write(content)
|
||||
Path(f.name).write_text(content)
|
||||
elif isinstance(content, bytes):
|
||||
with open(f.name, "wb") as fd:
|
||||
fd.write(content)
|
||||
Path(f.name).write_bytes(content)
|
||||
elif isinstance(content, io.IOBase):
|
||||
with open(f.name, "w") as fd:
|
||||
with Path(f.name).open("w") as fd:
|
||||
shutil.copyfileobj(content, fd)
|
||||
else:
|
||||
msg = f"Invalid content type: {type(content)}"
|
||||
@@ -191,13 +189,13 @@ def encrypt_file(
|
||||
# atomic copy of the encrypted file
|
||||
with NamedTemporaryFile(dir=folder, delete=False) as f2:
|
||||
shutil.copyfile(f.name, f2.name)
|
||||
os.rename(f2.name, secret_path)
|
||||
Path(f2.name).rename(secret_path)
|
||||
meta_path = secret_path.parent / "meta.json"
|
||||
with open(meta_path, "w") as f_meta:
|
||||
with meta_path.open("w") as f_meta:
|
||||
json.dump(meta, f_meta, indent=2)
|
||||
finally:
|
||||
with suppress(OSError):
|
||||
os.remove(f.name)
|
||||
Path(f.name).unlink()
|
||||
|
||||
|
||||
def decrypt_file(secret_path: Path) -> str:
|
||||
@@ -214,7 +212,7 @@ def get_meta(secret_path: Path) -> dict:
|
||||
meta_path = secret_path.parent / "meta.json"
|
||||
if not meta_path.exists():
|
||||
return {}
|
||||
with open(meta_path) as f:
|
||||
with meta_path.open() as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
@@ -233,7 +231,7 @@ def write_key(path: Path, publickey: str, overwrite: bool) -> None:
|
||||
|
||||
|
||||
def read_key(path: Path) -> str:
|
||||
with open(path / "key.json") as f:
|
||||
with Path(path / "key.json").open() as f:
|
||||
try:
|
||||
key = json.load(f)
|
||||
except json.JSONDecodeError as e:
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
@@ -20,7 +19,7 @@ def secret_name_type(arg_value: str) -> str:
|
||||
|
||||
|
||||
def public_or_private_age_key_type(arg_value: str) -> str:
|
||||
if os.path.isfile(arg_value):
|
||||
if Path(arg_value).is_file():
|
||||
arg_value = Path(arg_value).read_text().strip()
|
||||
if arg_value.startswith("age1"):
|
||||
return arg_value.strip()
|
||||
|
||||
Reference in New Issue
Block a user