diff --git a/src/composekit/__main__.py b/src/composekit/__main__.py index 17b5645..83f1a31 100644 --- a/src/composekit/__main__.py +++ b/src/composekit/__main__.py @@ -6,18 +6,18 @@ def _add_common(parser: argparse.ArgumentParser) -> None: - parser.add_argument( + _ = parser.add_argument( "-C", "--containers", help="Folder containing container definitions.", ) - parser.add_argument( + _ = parser.add_argument( "-c", "--config", action="append", help="Config file(s) to load (repeatable).", ) - parser.add_argument( + _ = parser.add_argument( "--commit", action="store_true", help="Commit the resulting changes to the git repository.", @@ -32,10 +32,10 @@ def build_parser() -> argparse.ArgumentParser: "generate", help="Create Docker Compose files." ) _add_common(gen) - gen.add_argument( + _ = gen.add_argument( "-o", "--composes", help="Folder to write per-service composes into." ) - gen.add_argument( + _ = gen.add_argument( "--output", help="Path of the aggregated main compose file." ) gen.set_defaults(func=generate.main) diff --git a/src/composekit/container.py b/src/composekit/container.py index 4359eae..74d7933 100644 --- a/src/composekit/container.py +++ b/src/composekit/container.py @@ -69,6 +69,7 @@ def load_containers(documents: Iterable[object]) -> list[Container]: data: dict[str, object] = {} for key, value in document.items(): + value: str | list[str] if not isinstance(key, str): raise TypeError("container keys must be strings") diff --git a/src/composekit/generate.py b/src/composekit/generate.py index 261307f..439d467 100755 --- a/src/composekit/generate.py +++ b/src/composekit/generate.py @@ -53,9 +53,12 @@ class Config(_Config): def get_folder_name(name: str, container: Container, config: Config) -> str: - folder = container.folder or name + folder = name + if isinstance(container.folder, str): + folder = container.folder + mode = config["capitalize_folder_name"] - if mode == "full" or (mode == "non_custom" and not container.folder): + if mode == "full" or (mode == "non_custom" and container.folder is None): folder = capitalize_name(folder) return folder @@ -109,7 +112,11 @@ def handle_volumes( if len(volume_segments) == 1: host_path = f"{bind_path}/{folder}" - volume_name = custom_name or volume_segments[0].rsplit("/", 1)[-1] + volume_name = ( + volume_segments[0].rsplit("/", 1)[-1] + if custom_name == "" + else custom_name + ) if volume_name in used_volumes: volume_name += str(used_volumes.count(volume_name) + 1) @@ -119,7 +126,7 @@ def handle_volumes( volume_segments = [host_path, volume_segments[0]] - if mount_option: + if mount_option is not None: volume_segments.append(mount_option) used_volumes.append(volume_segments[0].rsplit("/", 1)[-1]) @@ -147,7 +154,9 @@ def generate( "image": container.image, "hostname": name, "container_name": name, - "restart": container.restart or restart_policy, + "restart": ( + restart_policy if container.restart is None else container.restart + ), } for option in Container.fields(): @@ -170,7 +179,7 @@ def generate( case _: result[option] = value - if not container.network_mode: + if container.network_mode is None: result["networks"] = [network] return result @@ -242,11 +251,12 @@ def main(args: argparse.Namespace) -> None: containers = load_containers(yaml.safe_load_all(file)) for container in containers: - name = container.name or path.stem + name = path.stem if container.name is None else container.name + if name in used_names: number = str(used_names.count(name) + 1) container.name = name = f"{name}_{number}" - if container.folder: + if container.folder is not None: container.folder += number used_names.append(name) @@ -268,6 +278,6 @@ def main(args: argparse.Namespace) -> None: repo.git.add(".") staged_count = len(repo.index.diff(repo.head.commit)) if staged_count > 0: - repo.index.commit( + _ = repo.index.commit( f"chore(composes): update {staged_count} compose file(s)" ) diff --git a/src/composekit/sort.py b/src/composekit/sort.py index 10af990..4906472 100644 --- a/src/composekit/sort.py +++ b/src/composekit/sort.py @@ -41,8 +41,8 @@ async def process_file( yaml.dump_all(sorted_containers, file, sort_keys=False) if repo is not None: - repo.index.add(path) - repo.index.commit(f"chore({path.stem}): sort keys") + _ = repo.index.add(path) + _ = repo.index.commit(f"chore({path.stem}): sort keys") def main(args: argparse.Namespace) -> None: diff --git a/src/composekit/update.py b/src/composekit/update.py index adaf4b3..6062e6f 100755 --- a/src/composekit/update.py +++ b/src/composekit/update.py @@ -5,6 +5,7 @@ import logging import re import sys +from operator import itemgetter from pathlib import Path from typing import ClassVar @@ -39,11 +40,11 @@ class Config(_Config): def extract_version(version: str, pattern: str | None) -> str | None: - if not pattern: + if pattern is None: return version match = re.search(pattern, version) - if match and match.groups(): + if match is not None and len(match.groups()) > 0: return match.group(1) return None @@ -75,7 +76,7 @@ def parse_image(image: str) -> tuple[str | None, str | None, str, str] | None: def parse_version(version: str | None) -> Version | None: - if not version: + if version is None: return None try: @@ -111,7 +112,7 @@ async def find_versions( user: str | None, image: str, ) -> list[str]: - limit_config = options.get("limit") or config["limit"] + limit_config = options.get("limit", config["limit"]) limit = ( limit_config if isinstance(limit_config, int) @@ -125,16 +126,17 @@ async def find_versions( user = "library" if user is None else user username = options.get("username") password = options.get("password") - if tags := await list_tags( + tags = await list_tags( client, registry, f"{user}/{image}", username if isinstance(username, str) else None, password if isinstance(password, str) else None, - ): - return tags[-limit:] + ) + if len(tags) == 0: + raise Exception("No tags found.") - raise Exception("No tags found.") + return tags[-limit:] except Exception as e: logging.error(f"{full_image}: {e}") @@ -146,12 +148,14 @@ async def update( container: Container, client: httpx.AsyncClient, ) -> tuple[str, str, str] | None: - if not (result := parse_image(container.image)): + result = parse_image(container.image) + if result is None: return None registry, user, image, version = result full_image = "/".join(filter(None, [registry, user, image])) - registry = registry or str(config["default_registry"]) + if registry in (None, ""): + registry = str(config["default_registry"]) options = get_update_options(config, full_image, user, image) @@ -164,35 +168,33 @@ async def update( version_regex_config if isinstance(version_regex_config, str) else None ) - if not ( + if not isinstance( current_version := parse_version( extract_version(version, version_regex) - ) + ), + Version, ): logging.error( f"{full_image}: Could not parse the version '{version}'." ) return None - if not ( - raw_versions := await find_versions( - config, options, client, registry, user, image - ) - ): - return None + raw_versions = await find_versions( + config, options, client, registry, user, image + ) - versions = [ + versions: list[tuple[Version, str]] = [ (v, version) for version in raw_versions - if (v := parse_version(extract_version(version, version_regex))) + if isinstance( + v := parse_version(extract_version(version, version_regex)), + Version, + ) and v > current_version ] - if not versions: - return None - - newest_version = max(versions, key=lambda p: p[0], default=(None, None))[1] - if not newest_version: + newest_version = max(versions, key=itemgetter(0), default=(None, None))[1] + if newest_version is None: return None return full_image, image, newest_version @@ -209,7 +211,8 @@ async def process_file( containers = load_containers(yaml.safe_load_all(file)) for container in containers: - if not (result := await update(config, container, client)): + result = await update(config, container, client) + if result is None: continue full_image, image, newest_version = result @@ -224,8 +227,8 @@ async def process_file( ) if repo is not None: - repo.index.add(path) - repo.index.commit( + _ = repo.index.add(path) + _ = repo.index.commit( f"chore({path.stem}): update {image} to {newest_version}" ) diff --git a/src/composekit/utils/config.py b/src/composekit/utils/config.py index 1956543..058f762 100644 --- a/src/composekit/utils/config.py +++ b/src/composekit/utils/config.py @@ -25,8 +25,11 @@ def __setitem__(self, key: str, value: object) -> None: self.config[key] = value def __getitem__(self, key: str) -> object | None: - return ( - os.getenv(key.upper()) - or self.config.get(key.lower()) - or self.default_values.get(key) + sources = ( + os.getenv(key.upper()), + self.config.get(key.lower()), + self.default_values.get(key), ) + for value in sources: + if value is not None: + return value diff --git a/src/composekit/utils/git.py b/src/composekit/utils/git.py index 28aab1f..787e02e 100644 --- a/src/composekit/utils/git.py +++ b/src/composekit/utils/git.py @@ -10,6 +10,6 @@ def open_repo(reset: bool = True) -> Repo: repo = Repo(".", search_parent_directories=True) if reset: # Discard any changes - repo.index.reset(working_tree=True) + _ = repo.index.reset(working_tree=True) return repo diff --git a/src/composekit/utils/oci_api.py b/src/composekit/utils/oci_api.py index 43c7560..257f485 100644 --- a/src/composekit/utils/oci_api.py +++ b/src/composekit/utils/oci_api.py @@ -20,18 +20,18 @@ async def _list_tags_with_bearer_auth( params[k.strip()] = v.strip().strip('"') realm = params.pop("realm", None) - if not realm: + if not isinstance(realm, str): return [] request = await client.get(realm, params=params, auth=auth) - request.raise_for_status() - token_json = request.json() - token = token_json.get("token") or token_json.get("access_token") - if not token: + _ = request.raise_for_status() + token_json: dict[str, str] = request.json() + token = token_json.get("token", token_json.get("access_token")) + if not isinstance(token, str): raise RuntimeError("Token endpoint returned no token") r = await client.get(url, headers={"Authorization": f"Bearer {token}"}) - r.raise_for_status() + _ = r.raise_for_status() return r.json().get("tags", []) or [] @@ -42,7 +42,7 @@ async def list_tags( username: str | None = None, password: str | None = None, ) -> list[str]: - if not registry_host or registry_host == "docker.io": + if registry_host in (None, "", "docker.io"): registry_host = "index.docker.io" base = ( @@ -65,5 +65,5 @@ async def list_tags( www = r.headers.get("WWW-Authenticate", "") return await _list_tags_with_bearer_auth(client, url, www, auth) - r.raise_for_status() + _ = r.raise_for_status() return [] diff --git a/tests/test_generate.py b/tests/test_generate.py index 2d6241f..c64bd29 100644 --- a/tests/test_generate.py +++ b/tests/test_generate.py @@ -20,14 +20,17 @@ def make_mock_config(bind_path: str = "/bind") -> Config: + def get_config(key: str) -> object: + return { + "bind_path": bind_path, + "use_full_directory": True, + "capitalize_folder_name": False, + "restart_policy": "unless-stopped", + "network_name": "cloud", + }[key] + config = MagicMock(spec=Config) - config.__getitem__.side_effect = lambda key: { - "bind_path": bind_path, - "use_full_directory": True, - "capitalize_folder_name": False, - "restart_policy": "unless-stopped", - "network_name": "cloud", - }[key] + config.__getitem__.side_effect = get_config return config @@ -97,7 +100,7 @@ def test_main_handles_duplicate_containers_without_folder(self) -> None: composes = root / "composes" output = root / "docker-compose.yaml" containers.mkdir() - (containers / "container.yaml").write_text( + _ = (containers / "container.yaml").write_text( "image: nginx\n---\nimage: redis\n" ) @@ -111,7 +114,9 @@ def test_main_handles_duplicate_containers_without_folder(self) -> None: main(args) - compose = yaml.safe_load((composes / "container.yaml").read_text()) + compose: dict[str, dict[str, object]] = yaml.safe_load( + (composes / "container.yaml").read_text() + ) self.assertEqual( list(compose["services"].keys()), ["container", "container_2"] ) diff --git a/tests/test_update.py b/tests/test_update.py index 155bc9f..ea2ec09 100644 --- a/tests/test_update.py +++ b/tests/test_update.py @@ -74,8 +74,11 @@ def test_parse_version(self) -> None: class TestUpdate(unittest.IsolatedAsyncioTestCase): async def test_find_versions_mocked(self) -> None: + def get_config(key: str) -> object: + return {"limit": 2}[key] + config = MagicMock() - config.__getitem__.side_effect = lambda key: {"limit": 2}[key] + config.__getitem__.side_effect = get_config options: dict[str, object] = {} registry = None user = "user" @@ -94,15 +97,18 @@ async def test_find_versions_mocked(self) -> None: ) async def test_update_new_version(self) -> None: + def get_config(key: str) -> object: + return { + "default_registry": "docker.io", + "limit": 10, + "timeout": 5, + "user/image": {"update": True}, + "user": dict[str, object](), + "image": dict[str, object](), + }[key] + config = MagicMock() - config.__getitem__.side_effect = lambda key: { - "default_registry": "docker.io", - "limit": 10, - "timeout": 5, - "user/image": {"update": True}, - "user": dict[str, object](), - "image": dict[str, object](), - }[key] + config.__getitem__.side_effect = get_config container = Container(image="user/image:1.0.0") with patch( "composekit.update.find_versions", new_callable=AsyncMock @@ -117,15 +123,18 @@ async def test_update_new_version(self) -> None: self.assertEqual(image, "image") async def test_update_accepts_container(self) -> None: + def get_config(key: str) -> object: + return { + "default_registry": "docker.io", + "limit": 10, + "timeout": 5, + "user/image": {"update": True}, + "user": dict[str, object](), + "image": dict[str, object](), + }[key] + config = MagicMock() - config.__getitem__.side_effect = lambda key: { - "default_registry": "docker.io", - "limit": 10, - "timeout": 5, - "user/image": {"update": True}, - "user": dict[str, object](), - "image": dict[str, object](), - }[key] + config.__getitem__.side_effect = get_config container = Container(image="user/image:1.0.0") with patch( "composekit.update.find_versions", new_callable=AsyncMock @@ -137,13 +146,16 @@ async def test_update_accepts_container(self) -> None: self.assertEqual(result[2], "1.0.2") async def test_update_disabled(self) -> None: + def get_config(key: str) -> object: + return { + "default_registry": "docker.io", + "limit": 10, + "timeout": 5, + "user/image": {"update": False}, + }[key] + config = MagicMock() - config.__getitem__.side_effect = lambda key: { - "default_registry": "docker.io", - "limit": 10, - "timeout": 5, - "user/image": {"update": False}, - }[key] + config.__getitem__.side_effect = get_config container = Container(image="user/image:1.0.0") result = await update(config, container, AsyncMock()) self.assertIsNone(result)