Clean up type checking and handle unused values

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)

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")
 

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)"
  )

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:

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}"
  )
 

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

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

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 []

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"]
  )