Handle ruff warnings

gitopscli/__main__.py

@@ -14,10 +14,10 @@ def main() -> None:
         command.execute()
     except GitOpsException as ex:
         if verbose:
-            logging.exception(ex)
+            logging.exception(ex)  # noqa: TRY401
         else:
-            logging.error(ex)
-            logging.error("Provide verbose flag '-v' for more error details...")
+            logging.error(ex)  # noqa: TRY400
+            logging.error("Provide verbose flag '-v' for more error details...")  # noqa: TRY400
         sys.exit(1)
 
 

gitopscli/appconfig_api/app_tenant_config.py

@@ -17,10 +17,7 @@ class AppTenantConfig:
     dirty: bool = False
 
     def __post_init__(self) -> None:
-        if "config" in self.yaml:
-            self.tenant_config = self.yaml["config"]
-        else:
-            self.tenant_config = self.yaml
+        self.tenant_config = self.yaml.get("config", self.yaml)
         if "repository" not in self.tenant_config:
             raise GitOpsException("Cannot find key 'repository' in " + self.file_path)
         self.repo_url = str(self.tenant_config["repository"])
@@ -47,29 +44,28 @@ def __update_custom_app_config(self, desired_apps: dict[str, dict[str, Any]]) ->
                         )
                         del existing_application_value["customAppConfig"]
                         self.__set_dirty()
-                else:
-                    if (
-                        "customAppConfig" not in existing_application_value
-                        or existing_application_value["customAppConfig"] != desired_app_value["customAppConfig"]
-                    ):
-                        logging.info(
-                            "Updating customAppConfig in for %s in %s applications",
-                            existing_application_value,
-                            self.file_path,
-                        )
-                        existing_application_value["customAppConfig"] = desired_app_value["customAppConfig"]
-                        self.__set_dirty()
+                elif (
+                    "customAppConfig" not in existing_application_value
+                    or existing_application_value["customAppConfig"] != desired_app_value["customAppConfig"]
+                ):
+                    logging.info(
+                        "Updating customAppConfig in for %s in %s applications",
+                        existing_application_value,
+                        self.file_path,
+                    )
+                    existing_application_value["customAppConfig"] = desired_app_value["customAppConfig"]
+                    self.__set_dirty()
 
     def __add_new_applications(self, desired_apps: dict[str, Any]) -> None:
         for desired_app_name, desired_app_value in desired_apps.items():
-            if desired_app_name not in self.list_apps().keys():
+            if desired_app_name not in self.list_apps():
                 logging.info("Adding %s in %s applications", desired_app_name, self.file_path)
                 self.tenant_config["applications"][desired_app_name] = desired_app_value
                 self.__set_dirty()
 
     def __delete_removed_applications(self, desired_apps: dict[str, Any]) -> None:
-        for current_app in self.list_apps().keys():
-            if current_app not in desired_apps.keys():
+        for current_app in self.list_apps():
+            if current_app not in desired_apps:
                 logging.info("Removing %s from %s applications", current_app, self.file_path)
                 del self.tenant_config["applications"][current_app]
                 self.__set_dirty()
@@ -80,7 +76,7 @@ def __set_dirty(self) -> None:
 
 def __generate_config_from_tenant_repo(
     tenant_repo: GitRepo,
-) -> Any:  # TODO: supposed to be ruamel object than Any
+) -> Any:  # TODO: supposed to be ruamel object than Any # noqa: FIX002
     tenant_app_dirs = __get_all_tenant_applications_dirs(tenant_repo)
     tenant_config_template = f"""
     config:
@@ -103,19 +99,17 @@ def __generate_config_from_tenant_repo(
 
 def __get_all_tenant_applications_dirs(tenant_repo: GitRepo) -> set[str]:
     repo_dir = tenant_repo.get_full_file_path(".")
-    applist = {
+    return {
         name
         for name in os.listdir(repo_dir)
         if os.path.isdir(os.path.join(repo_dir, name)) and not name.startswith(".")
     }
-    return applist
 
 
 def __get_custom_config(appname: str, tenant_config_git_repo: GitRepo) -> Any:
     custom_config_path = tenant_config_git_repo.get_full_file_path(f"{appname}/.config.yaml")
     if os.path.exists(custom_config_path):
-        custom_config_content = yaml_file_load(custom_config_path)
-        return custom_config_content
+        return yaml_file_load(custom_config_path)
     return {}
 
 

gitopscli/appconfig_api/root_repo.py

@@ -31,7 +31,7 @@ def validate_tenant(self, tenant_config: AppTenantConfig) -> None:
         for tenant in self.tenants.values():
             if tenant.repo_url != tenant_config.repo_url:
                 apps_from_other_tenants.extend(tenant.list_apps().keys())
-        for app_name in tenant_config.list_apps().keys():
+        for app_name in tenant_config.list_apps():
             if app_name in apps_from_other_tenants:
                 raise GitOpsException(f"Application '{app_name}' already exists in a different repository")
 

gitopscli/commands/command_factory.py

@@ -10,7 +10,6 @@
 
 CommandArgs = (
     DeployCommand.Args
-    | DeployCommand.Args
     | AddPrCommentCommand.Args
     | CreatePreviewCommand.Args
     | CreatePrPreviewCommand.Args

gitopscli/commands/create_preview.py

@@ -175,7 +175,7 @@ def __create_preview_info_file(self, gitops_config: GitOpsConfig) -> None:
                 "routeHost": gitops_config.get_preview_host(preview_id),
                 "namespace": gitops_config.get_preview_namespace(preview_id),
             },
-            "/tmp/gitopscli-preview-info.yaml",
+            "/tmp/gitopscli-preview-info.yaml",  # noqa: S108
         )
 
     @staticmethod

gitopscli/commands/deploy.py

@@ -70,7 +70,7 @@ def execute(self) -> None:
                 git_repo_api.delete_branch(pr_branch)
 
         if self.__args.json:
-            print(json.dumps({"commits": [{"hash": h} for h in self.__commit_hashes]}, indent=4))
+            print(json.dumps({"commits": [{"hash": h} for h in self.__commit_hashes]}, indent=4))  # noqa: T201
 
     def __create_git_repo_api(self) -> GitRepoApi:
         return GitRepoApiFactory.create(self.__args, self.__args.organisation, self.__args.repository_name)
@@ -80,7 +80,7 @@ def __update_values(self, git_repo: GitRepo) -> dict[str, Any]:
         single_commit = args.single_commit or args.commit_message
         full_file_path = git_repo.get_full_file_path(args.file)
         updated_values = {}
-        for key, value in args.values.items():
+        for key, value in args.values.items():  # noqa: PD011
             try:
                 updated_value = update_yaml_file(full_file_path, key, value)
             except (FileNotFoundError, IsADirectoryError) as ex:
@@ -104,7 +104,7 @@ def __update_values(self, git_repo: GitRepo) -> dict[str, Any]:
             if args.commit_message:
                 message = args.commit_message
             elif len(updated_values) == 1:
-                key, value = list(updated_values.items())[0]
+                key, value = next(iter(updated_values.items()))
                 message = f"changed '{key}' to '{value}' in {args.file}"
             else:
                 updates_count = len(updated_values)

gitopscli/commands/sync_apps.py

@@ -34,19 +34,20 @@ def execute(self) -> None:
 def _sync_apps_command(args: SyncAppsCommand.Args) -> None:
     team_config_git_repo_api = GitRepoApiFactory.create(args, args.organisation, args.repository_name)
     root_config_git_repo_api = GitRepoApiFactory.create(args, args.root_organisation, args.root_repository_name)
-    with GitRepo(team_config_git_repo_api) as team_config_git_repo:
-        with GitRepo(root_config_git_repo_api) as root_config_git_repo:
-            __sync_apps(
-                team_config_git_repo,
-                root_config_git_repo,
-                args.git_user,
-                args.git_email,
-                args.git_author_name,
-                args.git_author_email,
-            )
+    with GitRepo(team_config_git_repo_api) as team_config_git_repo, GitRepo(
+        root_config_git_repo_api
+    ) as root_config_git_repo:
+        __sync_apps(
+            team_config_git_repo,
+            root_config_git_repo,
+            args.git_user,
+            args.git_email,
+            args.git_author_name,
+            args.git_author_email,
+        )
 
 
-# TODO: BETTER NAMES FOR STUFF HERE
+# TODO: BETTER NAMES FOR STUFF HERE # noqa: FIX002
 def __sync_apps(
     tenant_git_repo: GitRepo,
     root_git_repo: GitRepo,

gitopscli/commands/version.py

@@ -18,4 +18,4 @@ def execute(self) -> None:
         except importlib.metadata.PackageNotFoundError:
             # Handle the case where "gitopscli" is not installed
             version = None
-        print(f"GitOps CLI version {version}")
+        print(f"GitOps CLI version {version}")  # noqa: T201

gitopscli/git_api/bitbucket_git_repo_api_adapter.py

@@ -85,8 +85,8 @@ def create_pull_request(
     def merge_pull_request(
         self,
         pr_id: int,
-        merge_method: Literal["squash", "rebase", "merge"] = "merge",
-        merge_parameters: dict[str, Any] | None = None,
+        merge_method: Literal["squash", "rebase", "merge"] = "merge",  # noqa: ARG002
+        merge_parameters: dict[str, Any] | None = None,  # noqa: ARG002
     ) -> None:
         pull_request = self.__bitbucket.get_pullrequest(self.__organisation, self.__repository_name, pr_id)
         self.__bitbucket.merge_pull_request(

gitopscli/git_api/git_repo.py

@@ -133,7 +133,7 @@ def __get_repo(self) -> Repo:
 
     def __create_credentials_file(self, username: str, password: str) -> str:
         file_path = f"{self.__tmp_dir}/credentials.sh"
-        with open(file_path, "w", encoding=locale.getpreferredencoding(False)) as text_file:
+        with open(file_path, "w", encoding=locale.getpreferredencoding(do_setlocale=False)) as text_file:
             text_file.write("#!/bin/sh\n")
             text_file.write(f"echo username='{username}'\n")
             text_file.write(f"echo password='{password}'\n")

gitopscli/git_api/git_repo_api_logging_proxy.py

@@ -40,7 +40,7 @@ def merge_pull_request(
         self,
         pr_id: int,
         merge_method: Literal["squash", "rebase", "merge"] = "merge",
-        merge_parameters: dict[str, Any] | None = None,
+        merge_parameters: dict[str, Any] | None = None,  # noqa: ARG002
     ) -> None:
         logging.info("Merging pull request %s", pr_id)
         self.__api.merge_pull_request(pr_id, merge_method=merge_method)

gitopscli/git_api/github_git_repo_api_adapter.py

@@ -61,12 +61,17 @@ def merge_pull_request(
         self,
         pr_id: int,
         merge_method: Literal["squash", "rebase", "merge"] = "merge",
-        merge_parameters: dict[str, Any] | None = None,
+        merge_parameters: dict[str, Any] | None = None,  # noqa: ARG002
     ) -> None:
         pull_request = self.__get_pull_request(pr_id)
         pull_request.merge(merge_method=merge_method)
 
-    def add_pull_request_comment(self, pr_id: int, text: str, parent_id: int | None = None) -> None:
+    def add_pull_request_comment(
+        self,
+        pr_id: int,
+        text: str,
+        parent_id: int | None = None,  # noqa: ARG002
+    ) -> None:
         pull_request = self.__get_pull_request(pr_id)
         pull_request.create_issue_comment(text)
 

gitopscli/git_api/gitlab_git_repo_api_adapter.py

@@ -29,7 +29,7 @@ def __init__(
         except gitlab.exceptions.GitlabAuthenticationError as ex:
             raise GitOpsException("Bad Personal Access Token") from ex
         except gitlab.exceptions.GitlabGetError as ex:
-            if ex.response_code == 404:
+            if ex.response_code == 404:  # noqa: PLR2004
                 raise GitOpsException(f"Repository '{organisation}/{repository_name}' does not exist") from ex
             raise GitOpsException(f"Error getting repository: '{ex.error_message}'") from ex
 
@@ -82,7 +82,7 @@ def merge_pull_request(
                     merge_request.rebase(merge_parameters)
                     return
                 merge_request.merge(merge_parameters)
-                return
+                return  # noqa: TRY300
             except gitlab.exceptions.GitlabMRClosedError as ex:
                 # "Branch cannot be merged" error can occur if the server
                 # is still processing the merge request internally
@@ -96,7 +96,12 @@ def merge_pull_request(
                     raise GitOpsException("Error merging pull request: 'Branch cannot be merged'") from ex
                 time.sleep(2.5)
 
-    def add_pull_request_comment(self, pr_id: int, text: str, parent_id: int | None = None) -> None:
+    def add_pull_request_comment(
+        self,
+        pr_id: int,
+        text: str,
+        parent_id: int | None = None,  # noqa:ARG002
+    ) -> None:
         merge_request = self.__project.mergerequests.get(pr_id)
         merge_request.notes.create({"body": text})
 

gitopscli/gitops_config.py

@@ -3,7 +3,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from string import Template
-from typing import Any
+from typing import Any, ClassVar
 
 from gitopscli.gitops_exception import GitOpsException
 
@@ -19,7 +19,7 @@ class PreviewContext:
             preview_id: str
             git_hash: str
 
-        __VARIABLE_MAPPERS: dict[str, Callable[["GitOpsConfig.Replacement.PreviewContext"], str]] = {
+        __VARIABLE_MAPPERS: ClassVar[dict[str, Callable[["GitOpsConfig.Replacement.PreviewContext"], str]]] = {
             "GIT_HASH": lambda context: context.git_hash,
             "PREVIEW_HOST": lambda context: context.gitops_config.get_preview_host(context.preview_id),
             "PREVIEW_NAMESPACE": lambda context: context.gitops_config.get_preview_namespace(context.preview_id),
@@ -29,7 +29,7 @@ class PreviewContext:
             "PREVIEW_ID_HASH_SHORT": lambda context: GitOpsConfig.create_preview_id_hash_short(context.preview_id),
         }
 
-        def __init__(self, path: str, value_template: str):
+        def __init__(self, path: str, value_template: str) -> None:
             assert isinstance(path, str), "path of wrong type!"
             assert isinstance(value_template, str), "value_template of wrong type!"
 
@@ -118,8 +118,7 @@ def get_preview_host(self, preview_id: str) -> str:
         preview_host = preview_host.replace("${PREVIEW_ID_HASH}", self.create_preview_id_hash(preview_id))
         preview_host = preview_host.replace("${PREVIEW_ID_HASH_SHORT}", self.create_preview_id_hash_short(preview_id))
         preview_host = preview_host.replace("${PREVIEW_ID}", self.__sanitize(preview_id))
-        preview_host = preview_host.replace("${PREVIEW_NAMESPACE}", self.get_preview_namespace(preview_id))
-        return preview_host
+        return preview_host.replace("${PREVIEW_NAMESPACE}", self.get_preview_namespace(preview_id))
 
     def get_preview_namespace(self, preview_id: str) -> str:
         preview_namespace = self.preview_target_namespace_template
@@ -183,8 +182,7 @@ def __sanitize(preview_id: str, max_length: int | None = None) -> str:
         sanitized_preview_id = re.sub(r"-+", "-", sanitized_preview_id)
         if max_length is not None:
             sanitized_preview_id = sanitized_preview_id[0:max_length]
-        sanitized_preview_id = re.sub(r"-$", "", sanitized_preview_id)
-        return sanitized_preview_id
+        return re.sub(r"-$", "", sanitized_preview_id)
 
     def is_preview_template_equal_target(self) -> bool:
         return (
@@ -207,7 +205,7 @@ def from_yaml(yaml: Any) -> "GitOpsConfig":
 
 
 class _GitOpsConfigYamlParser:
-    def __init__(self, yaml: Any):
+    def __init__(self, yaml: Any) -> None:
         self.__yaml = yaml
 
     def __get_value(self, key: str) -> Any:

gitopscli/gitops_exception.py

@@ -1,2 +1,2 @@
-class GitOpsException(Exception):
+class GitOpsException(Exception):  # noqa: N818
     pass

gitopscli/io_api/tmp_dir.py

@@ -4,7 +4,7 @@
 
 
 def create_tmp_dir() -> str:
-    tmp_dir = f"/tmp/gitopscli/{uuid.uuid4()}"
+    tmp_dir = f"/tmp/gitopscli/{uuid.uuid4()}"  # noqa: S108
     os.makedirs(tmp_dir)
     return tmp_dir
 

gitopscli/io_api/yaml_util.py

@@ -10,20 +10,20 @@
 YAML_INSTANCE.preserve_quotes = True
 
 
-class YAMLException(Exception):
+class YAMLException(Exception):  # noqa: N818
     pass
 
 
 def yaml_file_load(file_path: str) -> Any:
-    with open(file_path, "r", encoding=locale.getpreferredencoding(False)) as stream:
+    with open(file_path, encoding=locale.getpreferredencoding(do_setlocale=False)) as stream:
         try:
             return YAML_INSTANCE.load(stream)
         except YAMLError as ex:
             raise YAMLException(f"Error parsing YAML file: {file_path}") from ex
 
 
 def yaml_file_dump(yaml: Any, file_path: str) -> None:
-    with open(file_path, "w+", encoding=locale.getpreferredencoding(False)) as stream:
+    with open(file_path, "w+", encoding=locale.getpreferredencoding(do_setlocale=False)) as stream:
         YAML_INSTANCE.dump(yaml, stream)
 
 
@@ -73,14 +73,13 @@ def merge_yaml_element(file_path: str, element_path: str, desired_value: Any) ->
             work_path = work_path[key]
 
     for key, value in desired_value.items():
-        if key in work_path:
-            if work_path[key] is not None:
-                value = {**work_path[key], **value}
+        if key in work_path and work_path[key] is not None:
+            value = {**work_path[key], **value}  # noqa: PLW2901
         work_path[key] = value
 
     # delete missing key:
     current = work_path.copy().items()
-    for key, value in current:
+    for key, _ in current:
         if key not in desired_value:
             del work_path[key]