use pathlib.iterdir() everywhere
This commit is contained in:
@@ -70,7 +70,7 @@ def complete_machines(
|
||||
if thread.is_alive():
|
||||
return iter([])
|
||||
|
||||
machines_dict = {name: "machine" for name in machines}
|
||||
machines_dict = dict.fromkeys(machines, "machine")
|
||||
return machines_dict
|
||||
|
||||
|
||||
@@ -113,7 +113,7 @@ def complete_services_for_machine(
|
||||
if thread.is_alive():
|
||||
return iter([])
|
||||
|
||||
services_dict = {name: "service" for name in services}
|
||||
services_dict = dict.fromkeys(services, "service")
|
||||
return services_dict
|
||||
|
||||
|
||||
@@ -155,7 +155,7 @@ def complete_backup_providers_for_machine(
|
||||
if thread.is_alive():
|
||||
return iter([])
|
||||
|
||||
providers_dict = {name: "provider" for name in providers}
|
||||
providers_dict = dict.fromkeys(providers, "provider")
|
||||
return providers_dict
|
||||
|
||||
|
||||
@@ -197,7 +197,7 @@ def complete_state_services_for_machine(
|
||||
if thread.is_alive():
|
||||
return iter([])
|
||||
|
||||
providers_dict = {name: "service" for name in providers}
|
||||
providers_dict = dict.fromkeys(providers, "service")
|
||||
return providers_dict
|
||||
|
||||
|
||||
@@ -219,7 +219,7 @@ def complete_secrets(
|
||||
|
||||
secrets = list_secrets(options.flake.path, options.pattern)
|
||||
|
||||
secrets_dict = {name: "secret" for name in secrets}
|
||||
secrets_dict = dict.fromkeys(secrets, "secret")
|
||||
return secrets_dict
|
||||
|
||||
|
||||
@@ -237,7 +237,7 @@ def complete_users(
|
||||
|
||||
users = list_users(Path(flake))
|
||||
|
||||
users_dict = {name: "user" for name in users}
|
||||
users_dict = dict.fromkeys(users, "user")
|
||||
return users_dict
|
||||
|
||||
|
||||
@@ -256,7 +256,7 @@ def complete_groups(
|
||||
groups_list = list_groups(Path(flake))
|
||||
groups = [group.name for group in groups_list]
|
||||
|
||||
groups_dict = {name: "group" for name in groups}
|
||||
groups_dict = dict.fromkeys(groups, "group")
|
||||
return groups_dict
|
||||
|
||||
|
||||
@@ -296,7 +296,7 @@ def complete_target_host(
|
||||
if thread.is_alive():
|
||||
return iter([])
|
||||
|
||||
providers_dict = {name: "target_host" for name in target_hosts}
|
||||
providers_dict = dict.fromkeys(target_hosts, "target_host")
|
||||
return providers_dict
|
||||
|
||||
|
||||
@@ -401,7 +401,7 @@ def complete_tags(
|
||||
if any(thread.is_alive() for thread in threads):
|
||||
return iter([])
|
||||
|
||||
providers_dict = {name: "tag" for name in tags}
|
||||
providers_dict = dict.fromkeys(tags, "tag")
|
||||
return providers_dict
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import os
|
||||
import shutil
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
@@ -27,9 +26,9 @@ def list_objects(path: Path, is_valid: Callable[[str], bool]) -> list[str]:
|
||||
objs: list[str] = []
|
||||
if not path.exists():
|
||||
return objs
|
||||
for f in os.listdir(path):
|
||||
if is_valid(f):
|
||||
objs.append(f)
|
||||
for f in path.iterdir():
|
||||
if is_valid(f.name):
|
||||
objs.append(f.name)
|
||||
return objs
|
||||
|
||||
|
||||
@@ -41,6 +40,6 @@ def remove_object(path: Path, name: str) -> list[Path]:
|
||||
except FileNotFoundError as e:
|
||||
msg = f"{name} not found in {path}"
|
||||
raise ClanError(msg) from e
|
||||
if not os.listdir(path):
|
||||
if not next(path.iterdir(), None):
|
||||
path.rmdir()
|
||||
return paths_to_commit
|
||||
|
||||
@@ -53,23 +53,23 @@ def list_groups(flake_dir: Path) -> list[Group]:
|
||||
if not groups_dir.exists():
|
||||
return groups
|
||||
|
||||
for group in os.listdir(groups_dir):
|
||||
for group in groups_dir.iterdir():
|
||||
group_folder = groups_dir / group
|
||||
if not group_folder.is_dir():
|
||||
continue
|
||||
machines_path = machines_folder(flake_dir, group)
|
||||
machines_path = machines_folder(flake_dir, group.name)
|
||||
machines = []
|
||||
if machines_path.is_dir():
|
||||
for f in machines_path.iterdir():
|
||||
if validate_hostname(f.name):
|
||||
machines.append(f.name)
|
||||
users_path = users_folder(flake_dir, group)
|
||||
users_path = users_folder(flake_dir, group.name)
|
||||
users = []
|
||||
if users_path.is_dir():
|
||||
for f in users_path.iterdir():
|
||||
if VALID_USER_NAME.match(f.name):
|
||||
users.append(f.name)
|
||||
groups.append(Group(flake_dir, group, machines, users))
|
||||
groups.append(Group(flake_dir, group.name, machines, users))
|
||||
return groups
|
||||
|
||||
|
||||
@@ -146,10 +146,10 @@ def remove_member(
|
||||
target.unlink()
|
||||
updated_files = [target]
|
||||
|
||||
if len(os.listdir(group_folder)) == 0:
|
||||
if next(group_folder.iterdir(), None) is None:
|
||||
group_folder.rmdir()
|
||||
|
||||
if len(os.listdir(group_folder.parent)) == 0:
|
||||
if next(group_folder.parent.iterdir(), None) is None:
|
||||
group_folder.parent.rmdir()
|
||||
|
||||
updated_files.extend(update_group_keys(flake_dir, group_name))
|
||||
|
||||
@@ -312,10 +312,10 @@ def disallow_member(group_folder: Path, name: str) -> list[Path]:
|
||||
raise ClanError(msg)
|
||||
target.unlink()
|
||||
|
||||
if len(os.listdir(group_folder)) == 0:
|
||||
if next(group_folder.iterdir(), None) is None:
|
||||
group_folder.rmdir()
|
||||
|
||||
if len(os.listdir(group_folder.parent)) == 0:
|
||||
if next(group_folder.parent.iterdir(), None) is None:
|
||||
group_folder.parent.rmdir()
|
||||
|
||||
return update_keys(target.parent.parent, collect_keys_for_path(group_folder.parent))
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
@@ -56,7 +55,7 @@ def remove_user(flake_dir: Path, name: str) -> None:
|
||||
# Remove the user from any group where it belonged:
|
||||
groups_dir = sops_groups_folder(flake_dir)
|
||||
if groups_dir.exists():
|
||||
for group in os.listdir(groups_dir):
|
||||
for group in groups_dir.iterdir():
|
||||
group_folder = groups_dir / group
|
||||
if not group_folder.is_dir():
|
||||
continue
|
||||
@@ -65,7 +64,7 @@ def remove_user(flake_dir: Path, name: str) -> None:
|
||||
continue
|
||||
log.info(f"Removing user {name} from group {group}")
|
||||
updated_paths.extend(
|
||||
groups.remove_member(flake_dir, group, groups.users_folder, name)
|
||||
groups.remove_member(flake_dir, group.name, groups.users_folder, name)
|
||||
)
|
||||
# Remove the user's key:
|
||||
updated_paths.extend(remove_object(sops_users_folder(flake_dir), name))
|
||||
|
||||
@@ -401,8 +401,8 @@ def test_groups(
|
||||
),
|
||||
)
|
||||
|
||||
groups = os.listdir(test_flake.path / "sops" / "groups")
|
||||
assert len(groups) == 0
|
||||
first_group = next((test_flake.path / "sops" / "groups").iterdir(), None)
|
||||
assert first_group is None
|
||||
|
||||
# Check if the symlink to the group was removed from our foo test secret:
|
||||
group_symlink = test_flake.path / "sops/secrets/foo/groups/group1"
|
||||
|
||||
Reference in New Issue
Block a user