Update pyrefly preset

pyproject.toml

@@ -76,9 +76,7 @@ quote-style = "double"
 [tool.pyrefly]
 project-includes = ["src", "tests"]
 min-severity = "info"
-preset = "strict"
-strict-callable-subtyping = true
-errors = { implicit-any = true, implicit-any-attribute = true, implicit-any-empty-container = true, implicit-any-parameter = true, implicit-any-type-argument = true, unannotated-parameter = true, unannotated-attribute = true, unannotated-return = true, unannotated-protocol-member = true, bad-override = true, bad-override-mutable-attribute = true, bad-override-param-name = true, bad-param-name-override = true, missing-override-decorator = true, inconsistent-inheritance = true, unused-ignore = true, unused-coroutine = true, redundant-cast = true, unreachable = true, redefinition = true, deprecated = true }
+preset = "all"
 
 [tool.commitizen]
 name = "cz_customize"

src/composekit/container.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass, fields
+
+Volume = str | dict[str, object]
+
+
+@dataclass
+class Container:
+    image: str
+    folder: str | None = None
+    name: str | None = None
+    restart: str | None = None
+    privileged: bool | None = None
+    network: str | None = None
+    network_mode: str | None = None
+    working_dir: str | None = None
+    command: str | None = None
+    entrypoint: str | None = None
+    user: str | None = None
+    shm_size: str | None = None
+    cap_add: list[str] | None = None
+    cap_drop: list[str] | None = None
+    group_add: list[str] | None = None
+    sysctls: list[str] | None = None
+    devices: list[str] | None = None
+    volumes: list[Volume] | None = None
+    tmpfs: list[str] | None = None
+    environment: list[str] | None = None
+    ports: list[str] | None = None
+    depends_on: list[str] | None = None
+    labels: list[str] | dict[str, str] | None = None
+    healthcheck: dict[str, object] | None = None
+
+    @classmethod
+    def fields(cls) -> list[str]:
+        return list(field.name for field in fields(cls))
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, object]) -> Container:
+        image = data.get("image")
+        if not isinstance(image, str):
+            raise TypeError("image must be a string")
+
+        container = cls(image=image)
+        for key, value in data.items():
+            if key == "image":
+                continue
+
+            if key in cls.fields():
+                setattr(container, key, value)
+
+        return container
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            key: getattr(self, key)
+            for key in self.fields()
+            if getattr(self, key) is not None
+        }
+
+
+def load_containers(documents: Iterable[object]) -> list[Container]:
+    containers: list[Container] = []
+    for document in documents:
+        if not isinstance(document, Mapping):
+            raise TypeError("container entry must be a mapping")
+
+        data: dict[str, object] = {}
+        for key, value in document.items():
+            if not isinstance(key, str):
+                raise TypeError("container keys must be strings")
+
+            data[key] = value
+
+        containers.append(Container.from_dict(data))
+
+    return containers

src/composekit/generate.py

@@ -5,11 +5,12 @@
 import shutil
 from collections.abc import Sequence
 from importlib.resources import files
-from typing import Any, ClassVar
+from typing import ClassVar
 
 try:
     import yaml
 
+    from composekit.container import Container, Volume, load_containers
     from composekit.utils import Config as _Config
     from composekit.utils import iter_container_files, open_repo
 except ImportError as err:
@@ -20,7 +21,7 @@
 
 class Config(_Config):
     config_paths = ("config/generate.yaml",)
-    default_values: ClassVar[dict[str, str | bool]] = {
+    default_values: ClassVar[dict[str, object]] = {
         "containers_folder": "containers",
         "composes_folder": "composes",
         "network_name": "cloud",
@@ -34,32 +35,6 @@ class Config(_Config):
     }
 
 
-OPTIONS = (
-    "folder",
-    "name",
-    "image",
-    "privileged",
-    "network",
-    "working_dir",
-    "command",
-    "network_mode",
-    "user",
-    "entrypoint",
-    "cap_add",
-    "cap_drop",
-    "sysctls",
-    "labels",
-    "devices",
-    "volumes",
-    "tmpfs",
-    "environment",
-    "group_add",
-    "depends_on",
-    "healthcheck",
-    "ports",
-    "shm_size",
-)
-
 MOUNT_OPTIONS = {
     "rw",
     "ro",
@@ -77,14 +52,10 @@ class Config(_Config):
 }
 
 
-def get_folder_name(
-    name: str, container: dict[str, Any], config: Config
-) -> str:
-    folder = str(container.get("folder", name))
+def get_folder_name(name: str, container: Container, config: Config) -> str:
+    folder = container.folder or name
     mode = config["capitalize_folder_name"]
-    if mode == "full" or (
-        mode == "non_custom" and not container.get("folder")
-    ):
+    if mode == "full" or (mode == "non_custom" and not container.folder):
         folder = capitalize_name(folder)
 
     return folder
@@ -94,7 +65,7 @@ def capitalize_name(name: str) -> str:
     return name[0].upper() + name[1:]
 
 
-def is_custom_bind(volume: str | dict[str, Any]) -> bool:
+def is_custom_bind(volume: Volume) -> bool:
     if isinstance(volume, dict):
         return True
 
@@ -112,13 +83,13 @@ def is_custom_bind(volume: str | dict[str, Any]) -> bool:
 def handle_volumes(
     config: Config,
     folder: str,
-    volumes: Sequence[str | dict[str, Any]],
+    volumes: Sequence[Volume],
     used_volumes: list[str],
-) -> list[str | dict[str, Any]]:
+) -> list[Volume]:
     bind_path = str(config["bind_path"])
     use_full_directory = bool(config["use_full_directory"])
     custom_binds = list(filter(is_custom_bind, volumes))
-    result: list[str | dict[str, Any]] = []
+    result: list[Volume] = []
 
     for volume in volumes:
         if isinstance(volume, dict):
@@ -164,26 +135,28 @@ def duplicate_entries(entries: list[str]) -> list[str]:
 
 
 def generate(
-    name: str, container: dict[str, Any], config: Config
-) -> dict[str, Any]:
+    name: str, container: Container, config: Config
+) -> dict[str, object]:
     folder = get_folder_name(name, container, config)
 
     restart_policy = str(config["restart_policy"])
     network = str(config["network_name"])
 
     used_volumes: list[str] = []
-    result: dict[str, Any] = {
-        "image": container.pop("image"),
+    result: dict[str, object] = {
+        "image": container.image,
         "hostname": name,
         "container_name": name,
-        "restart": container.get("restart", restart_policy),
+        "restart": container.restart or restart_policy,
     }
 
-    for option in OPTIONS:
-        if option not in container or option in ("folder", "name"):
+    for option in Container.fields():
+        if option in ("folder", "name", "image"):
             continue
 
-        value = container[option]
+        value = getattr(container, option)
+        if value is None:
+            continue
 
         match option:
             case "devices":
@@ -197,8 +170,8 @@ def generate(
             case _:
                 result[option] = value
 
-    if "network_mode" not in container:
-        result.setdefault("networks", []).append(network)
+    if not container.network_mode:
+        result["networks"] = [network]
 
     return result
 
@@ -244,7 +217,7 @@ def main(args: argparse.Namespace) -> None:
             gateway=gateway,
         )
     )
-    main_template["services"] = dict[str, Any]()
+    main_template["services"] = dict[str, object]()
 
     composes_template = (
         templates.joinpath("composes.yaml").read_text().lstrip()
@@ -260,21 +233,21 @@ def main(args: argparse.Namespace) -> None:
     for path in paths:
         used_names: list[str] = []
 
-        service: dict[str, dict[str, Any]] = yaml.safe_load(
+        service: dict[str, dict[str, object]] = yaml.safe_load(
             service_template.format(network=network)
         )
-        service["services"] = dict[str, Any]()
+        service["services"] = dict[str, object]()
 
         with open(path) as file:
-            containers: list[dict[str, Any]] = list(yaml.safe_load_all(file))
+            containers = load_containers(yaml.safe_load_all(file))
 
         for container in containers:
-            name = str(container.get("name", path.stem))
+            name = container.name or path.stem
             if name in used_names:
                 number = str(used_names.count(name) + 1)
-                container["name"] = name = f"{name}_{number}"
-                if container.get("folder"):
-                    container["folder"] += number
+                container.name = name = f"{name}_{number}"
+                if container.folder:
+                    container.folder += number
 
             used_names.append(name)
             service["services"][name] = generate(name, container, config)

src/composekit/sort.py

@@ -3,9 +3,9 @@
 import argparse
 import asyncio
 from pathlib import Path
-from typing import Any
 
-from .generate import OPTIONS, Config
+from .container import Container, load_containers
+from .generate import Config
 
 try:
     import yaml
@@ -24,17 +24,21 @@ async def process_file(
     git_lock: asyncio.Lock,
 ) -> None:
     with open(path) as file:
-        containers: list[dict[str, Any]] = list(yaml.safe_load_all(file))
+        containers = load_containers(yaml.safe_load_all(file))
 
-    for i, container in enumerate(containers):
-        sorted_dict = {k: container[k] for k in OPTIONS if k in container}
-        if list(container.keys()) == list(sorted_dict.keys()):
-            continue
+    sorted_containers: list[dict[str, object]] = []
+    changed = False
 
-        containers[i] = sorted_dict
+    for container in containers:
+        data = container.to_dict()
+        sorted_dict = {k: data[k] for k in Container.fields() if k in data}
+        sorted_containers.append(sorted_dict)
+        changed = changed or list(data.keys()) != list(sorted_dict.keys())
+
+    if changed:
         async with git_lock:
             with open(path, "w") as file:
-                yaml.dump_all(containers, file, sort_keys=False)
+                yaml.dump_all(sorted_containers, file, sort_keys=False)
 
             if repo is not None:
                 repo.index.add(path)