"""Pure Pydantic models for local storage specifications."""
from __future__ import annotations
import logging
from collections.abc import Sequence
from pathlib import Path
import yaml
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
[docs]
class LocalCollection(BaseModel):
"""A group of product specs sharing a directory template."""
directory: str
description: str | None = None
items: list = Field(default_factory=list)
[docs]
class LocalResourceSpec(BaseModel):
"""Root model for a local storage layout.
A single spec maps collection names to :class:`LocalCollection`
objects. Multiple specs can be merged via :meth:`merge` so that
different YAML files (e.g. per-project or per-workflow) combine
into one unified layout.
"""
name: str = "default"
description: str | None = None
collections: dict[str, LocalCollection] = Field(default_factory=dict)
source_file: Path | None = None
[docs]
@classmethod
def from_yaml(cls, path: str | Path) -> LocalResourceSpec:
"""Load from a YAML file.
Accepts either a top-level ``local:`` wrapper or a flat file
whose top-level key is ``collections:``.
Args:
path: Path to the YAML file.
Returns:
A :class:`LocalResourceSpec` instance.
"""
with open(path) as fh:
raw = yaml.safe_load(fh)
class_instance = cls.model_validate(raw.get("local", raw))
class_instance.source_file = Path(path)
return class_instance
[docs]
@classmethod
def merge(cls, specs: Sequence[LocalResourceSpec]) -> LocalResourceSpec:
"""Merge multiple local storage specs into one.
Later specs override collections with the same name. Items
within identically-named collections are combined (union).
Args:
specs: Sequence of specs to merge.
Returns:
A new :class:`LocalResourceSpec` with combined collections.
"""
merged_collections: dict[str, LocalCollection] = {}
name = "_".join(spec.name for spec in specs)
for spec in specs:
for coll_name, coll in spec.collections.items():
if coll_name in merged_collections:
existing = merged_collections[coll_name]
# Combine items (avoid duplicates, preserve order)
combined_items = list(existing.items)
for item in coll.items:
if item not in combined_items:
combined_items.append(item)
merged_collections[coll_name] = LocalCollection(
directory=coll.directory,
description=coll.description or existing.description,
items=combined_items,
)
else:
merged_collections[coll_name] = coll.model_copy(deep=True)
return cls(name=name, collections=merged_collections)