refactor: Make internal constants and classes private

Rename module-internal constants and classes with _ prefix:
- compose.py: SINGLE_PART, PUBLISHED_TARGET_PARTS, HOST_PUBLISHED_PARTS, MIN_VOLUME_PARTS
- logs.py: DIGEST_HEX_LENGTH
- traefik.py: LIST_VALUE_KEYS, MIN_ROUTER_PARTS, MIN_SERVICE_LABEL_PARTS,
  TraefikServiceSource, TRAEFIK_CONFIG_HEADER

These items are only used within their respective modules and should
not be part of the public API.
This commit is contained in:
Bas Nijholt
2025-12-15 20:33:48 -08:00
parent f8d88e6f97
commit 94aa58d380
3 changed files with 29 additions and 25 deletions

View File

@@ -18,10 +18,10 @@ if TYPE_CHECKING:
from .config import Config
# Port parsing constants
SINGLE_PART = 1
PUBLISHED_TARGET_PARTS = 2
HOST_PUBLISHED_PARTS = 3
MIN_VOLUME_PARTS = 2
_SINGLE_PART = 1
_PUBLISHED_TARGET_PARTS = 2
_HOST_PUBLISHED_PARTS = 3
_MIN_VOLUME_PARTS = 2
_VAR_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)(?::-(.*?))?\}")
@@ -93,12 +93,16 @@ def _parse_ports(raw: Any, env: dict[str, str]) -> list[PortMapping]: # noqa: P
published: int | None = None
target: int | None = None
if len(parts) == SINGLE_PART and parts[0].isdigit():
if len(parts) == _SINGLE_PART and parts[0].isdigit():
target = int(parts[0])
elif len(parts) == PUBLISHED_TARGET_PARTS and parts[0].isdigit() and parts[1].isdigit():
elif (
len(parts) == _PUBLISHED_TARGET_PARTS and parts[0].isdigit() and parts[1].isdigit()
):
published = int(parts[0])
target = int(parts[1])
elif len(parts) == HOST_PUBLISHED_PARTS and parts[-2].isdigit() and parts[-1].isdigit():
elif (
len(parts) == _HOST_PUBLISHED_PARTS and parts[-2].isdigit() and parts[-1].isdigit()
):
published = int(parts[-2])
target = int(parts[-1])
@@ -146,7 +150,7 @@ def _parse_volume_item(
if isinstance(item, str):
interpolated = _interpolate(item, env)
parts = interpolated.split(":")
if len(parts) >= MIN_VOLUME_PARTS:
if len(parts) >= _MIN_VOLUME_PARTS:
return _resolve_host_path(parts[0], compose_dir)
elif isinstance(item, dict) and item.get("type") == "bind":
source = item.get("source")

View File

@@ -20,7 +20,7 @@ if TYPE_CHECKING:
DEFAULT_LOG_PATH = xdg_config_home() / "compose-farm" / "dockerfarm-log.toml"
DIGEST_HEX_LENGTH = 64
_DIGEST_HEX_LENGTH = 64
@dataclass(frozen=True)
@@ -97,7 +97,7 @@ def _extract_image_fields(record: dict[str, Any]) -> tuple[str, str]:
or ""
)
if digest and not digest.startswith("sha256:") and len(digest) == DIGEST_HEX_LENGTH:
if digest and not digest.startswith("sha256:") and len(digest) == _DIGEST_HEX_LENGTH:
digest = f"sha256:{digest}"
return image, digest

View File

@@ -26,7 +26,7 @@ if TYPE_CHECKING:
@dataclass
class TraefikServiceSource:
class _TraefikServiceSource:
"""Source information to build an upstream for a Traefik service."""
traefik_service: str
@@ -38,9 +38,9 @@ class TraefikServiceSource:
scheme: str | None = None
LIST_VALUE_KEYS = {"entrypoints", "middlewares"}
MIN_ROUTER_PARTS = 3
MIN_SERVICE_LABEL_PARTS = 6
_LIST_VALUE_KEYS = {"entrypoints", "middlewares"}
_MIN_ROUTER_PARTS = 3
_MIN_SERVICE_LABEL_PARTS = 6
def _parse_value(key: str, raw_value: str) -> Any:
@@ -51,7 +51,7 @@ def _parse_value(key: str, raw_value: str) -> Any:
if value.isdigit():
return int(value)
last_segment = key.rsplit(".", 1)[-1]
if last_segment in LIST_VALUE_KEYS:
if last_segment in _LIST_VALUE_KEYS:
parts = [v.strip() for v in value.split(",")] if "," in value else [value]
return [part for part in parts if part]
return value
@@ -102,7 +102,7 @@ def _insert(root: dict[str, Any], key_path: list[str], value: Any) -> None: # n
current = container_list[list_index]
def _resolve_published_port(source: TraefikServiceSource) -> tuple[int | None, str | None]:
def _resolve_published_port(source: _TraefikServiceSource) -> tuple[int | None, str | None]:
"""Resolve host-published port for a Traefik service.
Returns (published_port, warning_message).
@@ -140,7 +140,7 @@ def _resolve_published_port(source: TraefikServiceSource) -> tuple[int | None, s
def _finalize_http_services(
dynamic: dict[str, Any],
sources: dict[str, TraefikServiceSource],
sources: dict[str, _TraefikServiceSource],
warnings: list[str],
) -> None:
for traefik_service, source in sources.items():
@@ -211,7 +211,7 @@ def _process_router_label(
if not key_without_prefix.startswith("http.routers."):
return
router_parts = key_without_prefix.split(".")
if len(router_parts) < MIN_ROUTER_PARTS:
if len(router_parts) < _MIN_ROUTER_PARTS:
return
router_name = router_parts[2]
router_remainder = router_parts[3:]
@@ -229,12 +229,12 @@ def _process_service_label(
host_address: str,
ports: list[PortMapping],
service_names: set[str],
sources: dict[str, TraefikServiceSource],
sources: dict[str, _TraefikServiceSource],
) -> None:
if not key_without_prefix.startswith("http.services."):
return
parts = key_without_prefix.split(".")
if len(parts) < MIN_SERVICE_LABEL_PARTS:
if len(parts) < _MIN_SERVICE_LABEL_PARTS:
return
traefik_service = parts[2]
service_names.add(traefik_service)
@@ -242,7 +242,7 @@ def _process_service_label(
source = sources.get(traefik_service)
if source is None:
source = TraefikServiceSource(
source = _TraefikServiceSource(
traefik_service=traefik_service,
stack=stack,
compose_service=compose_service,
@@ -267,7 +267,7 @@ def _process_service_labels(
host_address: str,
env: dict[str, str],
dynamic: dict[str, Any],
sources: dict[str, TraefikServiceSource],
sources: dict[str, _TraefikServiceSource],
warnings: list[str],
) -> None:
labels = normalize_labels(definition.get("labels"), env)
@@ -328,7 +328,7 @@ def generate_traefik_config(
"""
dynamic: dict[str, Any] = {}
warnings: list[str] = []
sources: dict[str, TraefikServiceSource] = {}
sources: dict[str, _TraefikServiceSource] = {}
# Determine Traefik's host from service assignment
traefik_host = None
@@ -366,7 +366,7 @@ def generate_traefik_config(
return dynamic, warnings
TRAEFIK_CONFIG_HEADER = """\
_TRAEFIK_CONFIG_HEADER = """\
# Auto-generated by compose-farm
# https://github.com/basnijholt/compose-farm
#
@@ -382,4 +382,4 @@ TRAEFIK_CONFIG_HEADER = """\
def render_traefik_config(dynamic: dict[str, Any]) -> str:
"""Render Traefik dynamic config as YAML with a header comment."""
body = yaml.safe_dump(dynamic, sort_keys=False)
return TRAEFIK_CONFIG_HEADER + body
return _TRAEFIK_CONFIG_HEADER + body