From 3712866b270350ae8fd0d8f461b5ac21d01dfafc Mon Sep 17 00:00:00 2001 From: Adrien Mannocci Date: Mon, 30 Dec 2024 15:15:08 +0100 Subject: [PATCH] chore: remove pylint references (#141) Signed-off-by: Adrien Mannocci --- scripts/build.py | 4 +- scripts/lint.py | 2 +- scripts/release.py | 49 +++++++++++---- terranova/binds.py | 4 -- terranova/commands/binds.py | 110 ++++++++++++++++++++++++++-------- terranova/commands/helpers.py | 17 ++++-- terranova/resources.py | 55 +++++++++++++---- terranova/utils.py | 5 +- 8 files changed, 184 insertions(+), 62 deletions(-) diff --git a/scripts/build.py b/scripts/build.py index 13c058f..5f7f8b5 100644 --- a/scripts/build.py +++ b/scripts/build.py @@ -50,7 +50,9 @@ def run() -> None: pyinstaller("terranova.spec", _out=sys.stdout, _err=sys.stderr) arch = platform.machine() arch = "amd64" if arch == "x86_64" else arch - Path("./dist/terranova").replace(Path(f"./dist/terranova-{version}-{system}-{arch}")) + Path("./dist/terranova").replace( + Path(f"./dist/terranova-{version}-{system}-{arch}") + ) elif system == "linux": # Use cross-build to build both amd64 and arm64 versions. cmd, env = container_backend() diff --git a/scripts/lint.py b/scripts/lint.py index 835b1f7..8aca24d 100644 --- a/scripts/lint.py +++ b/scripts/lint.py @@ -20,7 +20,7 @@ from sh import ErrorReturnCode -from scripts.utils import detect_git, detect_uv, detect_ruff, fatal +from scripts.utils import detect_git, detect_ruff, detect_uv, fatal def git_branch_delete(branch_name: str) -> None: diff --git a/scripts/release.py b/scripts/release.py index 64a6990..bf9b9cb 100644 --- a/scripts/release.py +++ b/scripts/release.py @@ -31,28 +31,42 @@ def __set_version(version: str) -> None: try: data = Constants.TERRANOVA_INIT_PATH.read_text() except Exception as err: - print(f"The `{Constants.TERRANOVA_INIT_PATH.as_posix()}` can't be read", file=sys.stderr) + print( + f"The `{Constants.TERRANOVA_INIT_PATH.as_posix()}` can't be read", + file=sys.stderr, + ) raise err - data = re.sub(r"__version__ = \"(.*)\"", f'__version__ = "{version}"', data, count=1) + data = re.sub( + r"__version__ = \"(.*)\"", f'__version__ = "{version}"', data, count=1 + ) try: Constants.TERRANOVA_INIT_PATH.write_text(data) except Exception as err: - print(f"The `{Constants.TERRANOVA_INIT_PATH.as_posix()}` file can't be written", file=sys.stderr) + print( + f"The `{Constants.TERRANOVA_INIT_PATH.as_posix()}` file can't be written", + file=sys.stderr, + ) raise err # Update project version try: data = Constants.PYPROJECT_PATH.read_text() except Exception as err: - print(f"The `{Constants.PYPROJECT_PATH.as_posix()}` can't be read", file=sys.stderr) + print( + f"The `{Constants.PYPROJECT_PATH.as_posix()}` can't be read", + file=sys.stderr, + ) raise err data = re.sub(r"version = \"(.+)\"", f'version = "{version}"', data, count=1) try: Constants.PYPROJECT_PATH.write_text(data) except Exception as err: - print(f"The `{Constants.PYPROJECT_PATH.as_posix()}` file can't be written", file=sys.stderr) + print( + f"The `{Constants.PYPROJECT_PATH.as_posix()}` file can't be written", + file=sys.stderr, + ) raise err @@ -60,9 +74,13 @@ def pre() -> None: # Ensure we have inputs release_version = os.getenv("RELEASE_VERSION") if not release_version: - return print("You must define `RELEASE_VERSION` environment variable.", file=sys.stderr) + return print( + "You must define `RELEASE_VERSION` environment variable.", file=sys.stderr + ) if not re.match(r"(\d){1,2}\.(\d){1,2}\.(\d){1,2}", release_version): - return print("The `RELEASE_VERSION` should match semver format.", file=sys.stderr) + return print( + "The `RELEASE_VERSION` should match semver format.", file=sys.stderr + ) # Create a new branch branch_name = f"release/v{release_version}" @@ -74,7 +92,14 @@ def pre() -> None: # Push release branch git("add", "--all", _out=sys.stdout, _err=sys.stderr) - git("commit", "-m", f"release: terranova v{release_version}", "--no-verify", _out=sys.stdout, _err=sys.stderr) + git( + "commit", + "-m", + f"release: terranova v{release_version}", + "--no-verify", + _out=sys.stdout, + _err=sys.stderr, + ) git("push", "origin", branch_name, _out=sys.stdout, _err=sys.stderr) # Create a PR @@ -100,7 +125,9 @@ def run() -> None: git("tag", release_version) git("push", "origin", release_version) except ErrorReturnCode: - return print(f"The release `v{release_version}` already exists.", file=sys.stderr) + return print( + f"The release `v{release_version}` already exists.", file=sys.stderr + ) # Create the release args = [ @@ -121,7 +148,9 @@ def post() -> None: # Ensure we have inputs next_version = os.getenv("NEXT_VERSION") if not next_version: - return print("You must define `NEXT_VERSION` environment variable", file=sys.stderr) + return print( + "You must define `NEXT_VERSION` environment variable", file=sys.stderr + ) if not re.match(r"(\d){1,2}\.(\d){1,2}\.(\d){1,2}-dev", next_version): return print("The `NEXT_VERSION` should match semver format.", file=sys.stderr) diff --git a/terranova/binds.py b/terranova/binds.py index eaa8d8a..8d468fc 100644 --- a/terranova/binds.py +++ b/terranova/binds.py @@ -35,7 +35,6 @@ from .utils import Constants, Log, SharedContext -# pylint: disable=R0903 class Bind: """Represents an abstract external bind.""" @@ -48,7 +47,6 @@ def binary_path(self) -> Path: Returns: path where binary is installed. """ - # pylint: disable=protected-access return Path(self._cmd._path) @contextmanager @@ -153,7 +151,6 @@ def is_allowed_env_var(env_var: str) -> bool: kwargs["_cwd"] = self.__work_dir return super()._exec_ctx(*args, **kwargs) - # pylint: disable=too-many-arguments,too-many-positional-arguments def init( self, backend_config: dict[str, str] | None = None, @@ -196,7 +193,6 @@ def fmt(self) -> None: """Reformat your configuration in the standard style.""" self._exec("fmt", _inherit=True) - # pylint: disable=redefined-builtin,too-many-positional-arguments def plan( self, compact_warnings: bool, diff --git a/terranova/commands/binds.py b/terranova/commands/binds.py index 9f25727..99bf30d 100644 --- a/terranova/commands/binds.py +++ b/terranova/commands/binds.py @@ -48,26 +48,40 @@ from terranova.utils import Constants, Log, SharedContext -# pylint: disable=R0914 @click.command("init") @click.argument("path", type=str, required=False) -@click.option("--migrate-state", help="Reconfigure a backend, and attempt to migrate any existing state.", is_flag=True) @click.option( - "--no-backend", help="Disable backend for this configuration and use what was previously instead.", is_flag=True + "--migrate-state", + help="Reconfigure a backend, and attempt to migrate any existing state.", + is_flag=True, +) +@click.option( + "--no-backend", + help="Disable backend for this configuration and use what was previously instead.", + is_flag=True, +) +@click.option( + "--reconfigure", + help="Reconfigure a backend, ignoring any saved configuration.", + is_flag=True, +) +@click.option( + "--upgrade", help="Install the latest module and provider versions.", is_flag=True ) -@click.option("--reconfigure", help="Reconfigure a backend, ignoring any saved configuration.", is_flag=True) -@click.option("--upgrade", help="Install the latest module and provider versions.", is_flag=True) @click.option( "--fail-at-end", help="If specified, only fail afterwards; allow all non-impacted projects to continue.", default=False, is_flag=True, ) -# pylint: disable-next=R0913,too-many-positional-arguments def init( - path: str | None, migrate_state: bool, no_backend: bool, reconfigure: bool, upgrade: bool, fail_at_end: bool + path: str | None, + migrate_state: bool, + no_backend: bool, + reconfigure: bool, + upgrade: bool, + fail_at_end: bool, ) -> None: - # pylint: disable=R0912 """Init resources manifest.""" # Find all resources manifests paths = resource_dirs(path) @@ -100,7 +114,9 @@ def init( target_dirname = os.path.dirname(dependency.target) os.symlink( os.path.relpath( - SharedContext.shared_dir().joinpath(dependency.source).as_posix(), + SharedContext.shared_dir() + .joinpath(dependency.source) + .as_posix(), full_path.joinpath(target_dirname).as_posix(), ), dependency.target, @@ -126,7 +142,9 @@ def init( # Mount terraform context terraform = mount_context(full_path, manifest) terraform.init( - backend_config={"key": os.path.relpath(full_path, SharedContext.resources_dir())}, + backend_config={ + "key": os.path.relpath(full_path, SharedContext.resources_dir()) + }, migrate_state=migrate_state, no_backend=no_backend, reconfigure=reconfigure, @@ -144,7 +162,9 @@ def init( @click.command("get") @click.argument("path", type=str, required=False) -@click.option("--selector", "selectors", type=SelectorType(), required=False, multiple=True) +@click.option( + "--selector", "selectors", type=SelectorType(), required=False, multiple=True +) def get(path: str | None, selectors: list[Selector] | None) -> None: """Display one or many resources.""" # Find all resources manifests @@ -221,7 +241,9 @@ def validate(path: str | None, fail_at_end: bool) -> None: # Report any errors if fail_at_end has been enabled if errors: - Log.fatal("The syntax is probably incorrect in one of the projects. See above for errors.") + Log.fatal( + "The syntax is probably incorrect in one of the projects. See above for errors." + ) @click.command("docs") @@ -240,7 +262,14 @@ def docs(docs_dir: Path) -> None: for file in files: if os.path.basename(file) == Constants.MANIFEST_FILE_NAME: jobs.append( - (Path(path), docs_dir.joinpath(os.path.relpath(path, SharedContext.resources_dir().as_posix()))) + ( + Path(path), + docs_dir.joinpath( + os.path.relpath( + path, SharedContext.resources_dir().as_posix() + ) + ), + ) ) # Clean docs dir @@ -260,10 +289,11 @@ def docs(docs_dir: Path) -> None: # Write documentation file rendering = tmpl.render({"manifest": manifest, "resources": resources}) formatted = mdformat.text(rendering) - target_path.with_suffix(".md").write_text(data=formatted, encoding=Constants.ENCODING_UTF_8) + target_path.with_suffix(".md").write_text( + data=formatted, encoding=Constants.ENCODING_UTF_8 + ) -# pylint: disable=redefined-builtin @click.command("plan") @click.argument("path", type=str, required=False) @click.option( @@ -271,9 +301,20 @@ def docs(docs_dir: Path) -> None: help="If Terraform produces any warnings that are not by errors, show them in a more compact that includes only the summary messages.", is_flag=True, ) -@click.option("--input/--no-input", help="Ask for input for variables if not directly set.", default=True) -@click.option("--no-color", help="If specified, output won't contain any color.", is_flag=True) -@click.option("--parallelism", help="Limit the number of parallel resource operations.", type=int, default=10) +@click.option( + "--input/--no-input", + help="Ask for input for variables if not directly set.", + default=True, +) +@click.option( + "--no-color", help="If specified, output won't contain any color.", is_flag=True +) +@click.option( + "--parallelism", + help="Limit the number of parallel resource operations.", + type=int, + default=10, +) @click.option( "--fail-at-end", help="If specified, only fail afterwards; allow all non-impacted projects to continue.", @@ -298,7 +339,6 @@ def docs(docs_dir: Path) -> None: type=click.Path(path_type=Path, dir_okay=False, writable=True), required=False, ) -# pylint: disable-next=R0913,too-many-positional-arguments def plan( path: str | None, compact_warnings: bool, @@ -343,10 +383,14 @@ def plan( args["out"] = path try: terraform.plan(**args) - execution_plan[rel_path] = b64encode(path.read_bytes()).decode(Constants.ENCODING_UTF_8) + execution_plan[rel_path] = b64encode(path.read_bytes()).decode( + Constants.ENCODING_UTF_8 + ) except ErrorReturnCode as plan_err: if plan_err.exit_code == 2: - execution_plan[rel_path] = b64encode(path.read_bytes()).decode(Constants.ENCODING_UTF_8) + execution_plan[rel_path] = b64encode( + path.read_bytes() + ).decode(Constants.ENCODING_UTF_8) raise plan_err else: terraform.plan(**args) @@ -383,7 +427,11 @@ def write_plan(): @click.command("apply") @click.argument("path_or_plan", type=str, required=False) -@click.option("--auto-approve", help="Skip interactive approval of plan before applying.", is_flag=True) +@click.option( + "--auto-approve", + help="Skip interactive approval of plan before applying.", + is_flag=True, +) @click.option("--target", help="Apply changes for specific target.", type=str) @click.option( "--fail-at-end", @@ -391,11 +439,15 @@ def write_plan(): default=False, is_flag=True, ) -def apply(path_or_plan: str | None, auto_approve: bool, target: str, fail_at_end: bool) -> None: +def apply( + path_or_plan: str | None, auto_approve: bool, target: str, fail_at_end: bool +) -> None: """Create or update resources.""" # Check if there is a plan to apply if path_or_plan.endswith("tnplan"): - execution_plan = json.loads(Path(path_or_plan).read_text(Constants.ENCODING_UTF_8)) + execution_plan = json.loads( + Path(path_or_plan).read_text(Constants.ENCODING_UTF_8) + ) paths = [] for rel_path in execution_plan.keys(): paths.append((SharedContext.resources_dir().joinpath(rel_path), rel_path)) @@ -421,7 +473,11 @@ def apply(path_or_plan: str | None, auto_approve: bool, target: str, fail_at_end with NamedTemporaryFile(prefix="terranova-") as file_descriptor: path = Path(file_descriptor.name) path.write_bytes(b64decode(execution_plan[rel_path])) - terraform.apply(plan=file_descriptor.name, auto_approve=auto_approve, target=target) + terraform.apply( + plan=file_descriptor.name, + auto_approve=auto_approve, + target=target, + ) else: terraform.apply(auto_approve=auto_approve, target=target) except ErrorReturnCode: @@ -547,7 +603,9 @@ def runbook(path: str, name: str) -> None: manifest = read_manifest(full_path) # Extract runbook - matching_runbooks = [rb for rb in manifest.runbooks if rb.name == name] if manifest.runbooks else [] + matching_runbooks = ( + [rb for rb in manifest.runbooks if rb.name == name] if manifest.runbooks else [] + ) if not matching_runbooks: Log.fatal("execute runbook", MissingRunbookError(name)) if len(matching_runbooks) > 1: diff --git a/terranova/commands/helpers.py b/terranova/commands/helpers.py index d2410f8..d12ae98 100644 --- a/terranova/commands/helpers.py +++ b/terranova/commands/helpers.py @@ -36,9 +36,10 @@ class SelectorType(click.ParamType): name = "selector" - # pylint: disable=R1710 @override - def convert(self, value, param: Parameter | None, ctx: click.Context | None) -> Selector: + def convert( + self, value, param: Parameter | None, ctx: click.Context | None + ) -> Selector: if not isinstance(value, str): self.fail(f"{value!r} isn't a valid selector", param, ctx) @@ -46,7 +47,6 @@ def convert(self, value, param: Parameter | None, ctx: click.Context | None) -> return Selector(name=data[0], value=None if len(data) == 1 else data[1]) -# pylint: disable=R1710 def read_manifest(path: Path) -> "ResourcesManifest": """ Read the resources manifest if possible. @@ -64,8 +64,9 @@ def read_manifest(path: Path) -> "ResourcesManifest": Log.fatal("read manifest", err) -# pylint: disable=R1710 -def discover_resources(path: Path, selectors: list[Selector] | None = None) -> list[Resource]: +def discover_resources( + path: Path, selectors: list[Selector] | None = None +) -> list[Resource]: """ Discover resources in every terraform configuration files. This function handle errors by logging and exiting. @@ -119,7 +120,11 @@ def resource_dirs(path: str | None) -> list[tuple[Path, str]]: return find_all_resource_dirs(resources_dir) -def mount_context(full_path: Path, manifest: ResourcesManifest | None = None, import_vars: bool = False) -> Terraform: +def mount_context( + full_path: Path, + manifest: ResourcesManifest | None = None, + import_vars: bool = False, +) -> Terraform: """Mount the terraform context by importing variables if needed.""" # Ensure manifest exists and can be read if not manifest: diff --git a/terranova/resources.py b/terranova/resources.py index 76ce145..12de1c3 100644 --- a/terranova/resources.py +++ b/terranova/resources.py @@ -112,7 +112,14 @@ def exec(self, path: str, workdir: Path) -> None: if self.workdir: workdir = workdir.joinpath(self.workdir) entrypoint = Command(self.entrypoint) - entrypoint(self.args, _env=env, _cwd=workdir, _in=sys.stdin, _out=sys.stdout, _err=sys.stderr) + entrypoint( + self.args, + _env=env, + _cwd=workdir, + _in=sys.stdin, + _out=sys.stdout, + _err=sys.stderr, + ) @dataclass_json @@ -153,7 +160,9 @@ def from_file(path: Path) -> "ResourcesManifest": if not os.access(path.as_posix(), os.R_OK): raise UnreadableManifestError(path) - with path.open(Constants.FILE_MODE_READ, encoding=Constants.ENCODING_UTF_8) as file_descriptor: + with path.open( + Constants.FILE_MODE_READ, encoding=Constants.ENCODING_UTF_8 + ) as file_descriptor: try: data = yaml.safe_load(file_descriptor) except yaml.YAMLError as err: @@ -164,9 +173,13 @@ def from_file(path: Path) -> "ResourcesManifest": # Get configuration schema try: - schema = pkgutil.get_data(__name__, f"schemas/manifest_schema_v{version}.json") + schema = pkgutil.get_data( + __name__, f"schemas/manifest_schema_v{version}.json" + ) if not schema: - raise FileNotFoundError(f"The schema `schemas/manifest_schema_v{version}.json` can't be found") + raise FileNotFoundError( + f"The schema `schemas/manifest_schema_v{version}.json` can't be found" + ) except FileNotFoundError as err: raise VersionManifestError(version) from err schema = json.loads(schema) @@ -177,7 +190,6 @@ def from_file(path: Path) -> "ResourcesManifest": except ValidationError as err: raise InvalidManifestError(path) from err # noinspection PyUnresolvedReferences - # pylint: disable=E1101 return ResourcesManifest.from_dict(data) # type: ignore[attr-defined] @@ -225,10 +237,14 @@ class ResourcesFinder: __RESOURCE_PATTERN: Pattern = re.compile( r"""(/\*(?P[@\S\s\n]*?)\*/\n)?(?Presource|data) \"(?P\w+)\" \"(?P[a-zA-Z0-9_-]+)\"""" ) - __RESOURCE_ATTR_PATTERN: Pattern = re.compile(r"""@(?P\S+)\s+(?P.+)""") + __RESOURCE_ATTR_PATTERN: Pattern = re.compile( + r"""@(?P\S+)\s+(?P.+)""" + ) @staticmethod - def find_in_dir(path: Path, selectors: list[Selector] | None = None) -> list[Resource]: + def find_in_dir( + path: Path, selectors: list[Selector] | None = None + ) -> list[Resource]: """ Find all resources in directory. @@ -244,9 +260,10 @@ def find_in_dir(path: Path, selectors: list[Selector] | None = None) -> list[Res resources += ResourcesFinder.find_in_file(file, selectors) return resources - # pylint: disable=R0914 @staticmethod - def find_in_file(path: Path, selectors: list[Selector] | None = None) -> list[Resource]: + def find_in_file( + path: Path, selectors: list[Selector] | None = None + ) -> list[Resource]: """ Find all resources in a file. @@ -275,9 +292,18 @@ def find_in_file(path: Path, selectors: list[Selector] | None = None) -> list[Re resolution="Add metadata for the above resource.", ) - _, maybe_resource_attrs, resource_block_type, resource_type, resource_name = resource_match + ( + _, + maybe_resource_attrs, + resource_block_type, + resource_type, + resource_name, + ) = resource_match maybe_resource_attrs = maybe_resource_attrs.strip() - if not maybe_resource_attrs and resource_block_type == ResourceBlockType.RESOURCE: + if ( + not maybe_resource_attrs + and resource_block_type == ResourceBlockType.RESOURCE + ): raise InvalidResourcesError( cause=f"The resource `{resource_block_type}:{resource_type}:{resource_name}` at `{path.as_posix()}` isn't describe.", resolution="Add metadata for the above resource.", @@ -290,7 +316,12 @@ def find_in_file(path: Path, selectors: list[Selector] | None = None) -> list[Re if attr_match: name, value = attr_match.groups() attrs[name].append(value) - resource = Resource(block_type=resource_block_type, name=resource_name, type=resource_type, attrs=attrs) + resource = Resource( + block_type=resource_block_type, + name=resource_name, + type=resource_type, + attrs=attrs, + ) # Filter resource by selector match = True diff --git a/terranova/utils.py b/terranova/utils.py index 621f262..58928fa 100644 --- a/terranova/utils.py +++ b/terranova/utils.py @@ -29,7 +29,6 @@ class Constants: """All constants""" - # pylint: disable=R0903 CTX_CONF_DIR: Final[str] = "ctx_conf_dir" CTX_CONSOLE: Final[str] = "ctx_console" CTX_DEBUG: Final[str] = "ctx_debug" @@ -149,7 +148,9 @@ def failure(cls, msgs: str | list[str], err: Exception | None = None) -> None: err_console.print(f" Details: {err}") @classmethod - def fatal(cls, msgs: str | list[str], err: Exception | None = None, raise_exit: int = 1) -> NoReturn: + def fatal( + cls, msgs: str | list[str], err: Exception | None = None, raise_exit: int = 1 + ) -> NoReturn: """Log a failure and exit.""" Log.failure(msgs, err) raise Exit(code=raise_exit)