2.1 Profile resolver service (el-1nj)
This commit is contained in:
@@ -0,0 +1,251 @@
|
||||
"""Profile resolver service for recursive ordered include resolution.
|
||||
|
||||
Provides deterministic merge rules, save-independent cycle protection,
|
||||
and resolved output structures for env vars, runtime hints, mounts,
|
||||
file trees, and override metadata.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from src.models.config_include import ConfigInclude
|
||||
from src.models.config_mount import ConfigMount
|
||||
from src.models.config_profile import ConfigProfile
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResolvedMount:
|
||||
"""A resolved mount with merged file tree and final mode."""
|
||||
|
||||
target_path: str
|
||||
mode: str # "ro" or "rw"
|
||||
files: dict[str, str] = field(default_factory=dict)
|
||||
"""Relative file paths to UTF-8 text content."""
|
||||
overridden_files: dict[str, list[str]] = field(default_factory=dict)
|
||||
"""Map of relative file path to list of profile names that contributed
|
||||
(latest is the winner)."""
|
||||
mode_overridden_by: str | None = None
|
||||
"""Name of the profile that set the final mode, if different from first."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResolvedRuntimeHints:
|
||||
"""Resolved runtime hints from profile layers."""
|
||||
|
||||
start_command: str | None = None
|
||||
working_directory: str | None = None
|
||||
port: int | None = None
|
||||
overridden_hints: dict[str, str] = field(default_factory=dict)
|
||||
"""Map of hint key to profile name that provided the winning value."""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResolvedProfileOutput:
|
||||
"""Complete resolved output for a config profile."""
|
||||
|
||||
profile_id: uuid.UUID
|
||||
profile_name: str
|
||||
environment_variables: dict[str, str] = field(default_factory=dict)
|
||||
"""Final merged env vars (later layers win)."""
|
||||
env_var_sources: dict[str, list[str]] = field(default_factory=dict)
|
||||
"""Map of env var key to ordered list of contributing profile names
|
||||
(latest is the winner)."""
|
||||
runtime_hints: ResolvedRuntimeHints = field(
|
||||
default_factory=lambda: ResolvedRuntimeHints()
|
||||
)
|
||||
mounts: dict[str, ResolvedMount] = field(default_factory=dict)
|
||||
"""Map of target_path to ResolvedMount."""
|
||||
resolution_order: list[str] = field(default_factory=list)
|
||||
"""Ordered list of profile names as they were resolved."""
|
||||
cycle_detected: bool = False
|
||||
cycle_path: list[str] | None = None
|
||||
|
||||
|
||||
class ProfileResolutionError(Exception):
|
||||
"""Raised when profile resolution fails."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class ProfileCycleError(ProfileResolutionError):
|
||||
"""Raised when a cycle is detected during profile resolution."""
|
||||
|
||||
def __init__(self, cycle_path: list[str]) -> None:
|
||||
self.cycle_path = cycle_path
|
||||
path_str = " -> ".join(cycle_path)
|
||||
super().__init__(f"Profile include cycle detected: {path_str}")
|
||||
|
||||
|
||||
def _merge_env_vars(
|
||||
current: dict[str, str],
|
||||
sources: dict[str, list[str]],
|
||||
profile: ConfigProfile,
|
||||
) -> None:
|
||||
"""Merge a profile's env vars into the current dict, tracking sources."""
|
||||
if not profile.environment_variables:
|
||||
return
|
||||
for key, value in profile.environment_variables.items():
|
||||
current[key] = value
|
||||
if key not in sources:
|
||||
sources[key] = []
|
||||
sources[key].append(profile.name)
|
||||
|
||||
|
||||
def _merge_runtime_hints(
|
||||
hints: ResolvedRuntimeHints,
|
||||
profile: ConfigProfile,
|
||||
) -> None:
|
||||
"""Merge a profile's runtime hints, tracking overrides."""
|
||||
if profile.start_command is not None:
|
||||
hints.start_command = profile.start_command
|
||||
hints.overridden_hints["start_command"] = profile.name
|
||||
if profile.working_directory is not None:
|
||||
hints.working_directory = profile.working_directory
|
||||
hints.overridden_hints["working_directory"] = profile.name
|
||||
if profile.port is not None:
|
||||
hints.port = profile.port
|
||||
hints.overridden_hints["port"] = profile.name
|
||||
|
||||
|
||||
def _merge_mounts(
|
||||
mounts: dict[str, ResolvedMount],
|
||||
profile_mounts: list[ConfigMount],
|
||||
profile: ConfigProfile,
|
||||
) -> None:
|
||||
"""Merge a profile's mounts into the current mounts dict."""
|
||||
for mount in profile_mounts:
|
||||
target = mount.target_path
|
||||
if target not in mounts:
|
||||
mounts[target] = ResolvedMount(
|
||||
target_path=target,
|
||||
mode=mount.mode,
|
||||
files={},
|
||||
overridden_files={},
|
||||
)
|
||||
resolved = mounts[target]
|
||||
|
||||
# Mode override: later wins
|
||||
if resolved.mode != mount.mode:
|
||||
resolved.mode = mount.mode
|
||||
resolved.mode_overridden_by = profile.name
|
||||
|
||||
# File tree merge: later wins for same relative path
|
||||
if mount.files:
|
||||
for rel_path, content in mount.files.items():
|
||||
if rel_path not in resolved.files:
|
||||
resolved.overridden_files[rel_path] = []
|
||||
else:
|
||||
if rel_path not in resolved.overridden_files:
|
||||
resolved.overridden_files[rel_path] = []
|
||||
resolved.overridden_files[rel_path].append(profile.name)
|
||||
resolved.files[rel_path] = content
|
||||
|
||||
|
||||
def _resolve_profile_recursive(
|
||||
profile: ConfigProfile,
|
||||
visited: set[uuid.UUID],
|
||||
path: list[str],
|
||||
resolution_order: list[str],
|
||||
env_vars: dict[str, str],
|
||||
env_var_sources: dict[str, list[str]],
|
||||
runtime_hints: ResolvedRuntimeHints,
|
||||
mounts: dict[str, ResolvedMount],
|
||||
) -> None:
|
||||
"""Recursively resolve a profile and its includes.
|
||||
|
||||
Args:
|
||||
profile: The profile to resolve
|
||||
visited: Set of already-resolved profile IDs to avoid duplicates
|
||||
path: Current recursion path for cycle detection
|
||||
resolution_order: Ordered list of profile names being resolved
|
||||
env_vars: Accumulated environment variables
|
||||
env_var_sources: Tracking of which profiles contributed each env var
|
||||
runtime_hints: Accumulated runtime hints
|
||||
mounts: Accumulated mounts
|
||||
|
||||
Raises:
|
||||
ProfileCycleError: If a cycle is detected
|
||||
"""
|
||||
if profile.name in path:
|
||||
# Cycle detected
|
||||
cycle_start = path.index(profile.name)
|
||||
cycle_path = path[cycle_start:] + [profile.name]
|
||||
raise ProfileCycleError(cycle_path)
|
||||
|
||||
if profile.id in visited:
|
||||
# Already resolved in another branch (diamond graph)
|
||||
return
|
||||
|
||||
visited.add(profile.id)
|
||||
path.append(profile.name)
|
||||
resolution_order.append(profile.name)
|
||||
|
||||
# Resolve includes first (in order)
|
||||
includes: list[ConfigInclude] = list(profile.includes)
|
||||
includes.sort(key=lambda inc: inc.order_index)
|
||||
for include in includes:
|
||||
included_profile = include.included_profile
|
||||
if included_profile is not None:
|
||||
_resolve_profile_recursive(
|
||||
included_profile,
|
||||
visited,
|
||||
path,
|
||||
resolution_order,
|
||||
env_vars,
|
||||
env_var_sources,
|
||||
runtime_hints,
|
||||
mounts,
|
||||
)
|
||||
|
||||
# Apply this profile's values (later layers win)
|
||||
_merge_env_vars(env_vars, env_var_sources, profile)
|
||||
_merge_runtime_hints(runtime_hints, profile)
|
||||
_merge_mounts(mounts, list(profile.mounts), profile)
|
||||
|
||||
path.pop()
|
||||
|
||||
|
||||
def resolve_profile(profile: ConfigProfile) -> ResolvedProfileOutput:
|
||||
"""Resolve a config profile with all its includes.
|
||||
|
||||
Processes included profiles in configured order, then applies the
|
||||
selected profile itself. Later layers override earlier layers.
|
||||
|
||||
Args:
|
||||
profile: The root profile to resolve
|
||||
|
||||
Returns:
|
||||
ResolvedProfileOutput with merged env vars, runtime hints, mounts,
|
||||
and override metadata
|
||||
|
||||
Raises:
|
||||
ProfileCycleError: If a cycle is detected in the include graph
|
||||
"""
|
||||
env_vars: dict[str, str] = {}
|
||||
env_var_sources: dict[str, list[str]] = {}
|
||||
runtime_hints = ResolvedRuntimeHints()
|
||||
mounts: dict[str, ResolvedMount] = {}
|
||||
resolution_order: list[str] = []
|
||||
|
||||
_resolve_profile_recursive(
|
||||
profile,
|
||||
set(),
|
||||
[],
|
||||
resolution_order,
|
||||
env_vars,
|
||||
env_var_sources,
|
||||
runtime_hints,
|
||||
mounts,
|
||||
)
|
||||
|
||||
return ResolvedProfileOutput(
|
||||
profile_id=profile.id,
|
||||
profile_name=profile.name,
|
||||
environment_variables=env_vars,
|
||||
env_var_sources=env_var_sources,
|
||||
runtime_hints=runtime_hints,
|
||||
mounts=mounts,
|
||||
resolution_order=resolution_order,
|
||||
)
|
||||
Reference in New Issue
Block a user