Skip to content

Output

Utilities for collecting and serializing manifests.

Manifest Collection

Manifest dataclass

A collection of Kubernetes resources.

Example

manifest = Manifest() manifest.add({"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "test"}}) manifest.add(build_deployment(spec)) print(manifest.to_yaml()) manifest.write("manifests/app.yaml")

Source code in src/k8smith/output/manifest.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@dataclass
class Manifest:
    """A collection of Kubernetes resources.

    Example:
        >>> manifest = Manifest()
        >>> manifest.add({"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "test"}})
        >>> manifest.add(build_deployment(spec))
        >>> print(manifest.to_yaml())
        >>> manifest.write("manifests/app.yaml")
    """

    resources: list[dict] = field(default_factory=list)

    def add(self, resource: dict) -> Manifest:
        """Add a resource to the manifest.

        Args:
            resource: A Kubernetes resource dict

        Returns:
            Self for chaining
        """
        self.resources.append(resource)
        return self

    def add_all(self, resources: list[dict]) -> Manifest:
        """Add multiple resources to the manifest.

        Args:
            resources: List of Kubernetes resource dicts

        Returns:
            Self for chaining
        """
        self.resources.extend(resources)
        return self

    def to_yaml(self) -> str:
        """Serialize to YAML string.

        Returns:
            YAML string with --- document separators
        """
        return dump(self.resources)

    def write(self, path: str | Path) -> None:
        """Write manifest to a file.

        Args:
            path: File path to write to
        """
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(self.to_yaml())

    @classmethod
    def from_yaml(cls, yaml_str: str) -> Manifest:
        """Load manifest from a YAML string.

        Args:
            yaml_str: YAML string with one or more documents

        Returns:
            Manifest containing the loaded resources
        """
        return cls(resources=load(yaml_str))

    @classmethod
    def from_file(cls, path: str | Path) -> Manifest:
        """Load manifest from a file.

        Args:
            path: Path to YAML file

        Returns:
            Manifest containing the loaded resources
        """
        path = Path(path)
        return cls.from_yaml(path.read_text())

    def filter(
        self,
        *,
        kind: str | None = None,
        namespace: str | None = None,
        name: str | None = None,
        labels: dict[str, str] | None = None,
    ) -> Manifest:
        """Filter resources by criteria.

        Args:
            kind: Filter by resource kind (e.g., "Deployment")
            namespace: Filter by namespace
            name: Filter by resource name
            labels: Filter by labels (all must match)

        Returns:
            New Manifest with filtered resources
        """
        filtered = []
        for r in self.resources:
            if kind and r.get("kind") != kind:
                continue
            if namespace and r.get("metadata", {}).get("namespace") != namespace:
                continue
            if name and r.get("metadata", {}).get("name") != name:
                continue
            if labels:
                resource_labels = r.get("metadata", {}).get("labels", {})
                if not all(resource_labels.get(k) == v for k, v in labels.items()):
                    continue
            filtered.append(r)
        return Manifest(resources=filtered)

    def __len__(self) -> int:
        """Return the number of resources in the manifest."""
        return len(self.resources)

    def __iter__(self) -> Iterator[dict[str, Any]]:
        """Iterate over resources."""
        return iter(self.resources)

    def __getitem__(self, index: int) -> dict:
        """Get a resource by index."""
        return self.resources[index]

add(resource)

Add a resource to the manifest.

Parameters:

Name Type Description Default
resource dict

A Kubernetes resource dict

 required

Returns:

Type Description
Manifest

Self for chaining
Source code in src/k8smith/output/manifest.py 
27
28
29
30
31
32
33
34
35
36
37
def add(self, resource: dict) -> Manifest:
    """Add a resource to the manifest.

    Args:
        resource: A Kubernetes resource dict

    Returns:
        Self for chaining
    """
    self.resources.append(resource)
    return self

to_yaml()

Serialize to YAML string.

Returns:

Type Description
str

YAML string with --- document separators
Source code in src/k8smith/output/manifest.py 
51
52
53
54
55
56
57
def to_yaml(self) -> str:
    """Serialize to YAML string.

    Returns:
        YAML string with --- document separators
    """
    return dump(self.resources)

YAML Serialization

dump(resources)

Dump resources to a YAML string with document separators.

Parameters:

Name Type Description Default
resources list[dict]

List of Kubernetes resource dicts

 required

Returns:

Type Description
str

YAML string with --- document separators
Example

resources = [{"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "test"}}] print(dump(resources)) apiVersion: v1 kind: Namespace metadata: name: test

Source code in src/k8smith/output/yaml.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def dump(resources: list[dict]) -> str:
    """Dump resources to a YAML string with document separators.

    Args:
        resources: List of Kubernetes resource dicts

    Returns:
        YAML string with --- document separators

    Example:
        >>> resources = [{"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "test"}}]
        >>> print(dump(resources))
        apiVersion: v1
        kind: Namespace
        metadata:
          name: test
    """
    if not resources:
        return ""

    docs = []
    for resource in resources:
        doc = yaml.dump(
            resource,
            Dumper=KubernetesYamlDumper,
            default_flow_style=False,
            sort_keys=False,
            allow_unicode=True,
        )
        docs.append(doc.rstrip())

    return "\n---\n".join(docs) + "\n"

dump_one(resource)

Dump a single resource to YAML.

Parameters:

Name Type Description Default
resource dict

A Kubernetes resource dict

 required

Returns:

Type Description
str

YAML string
Source code in src/k8smith/output/yaml.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def dump_one(resource: dict) -> str:
    """Dump a single resource to YAML.

    Args:
        resource: A Kubernetes resource dict

    Returns:
        YAML string
    """
    return yaml.dump(
        resource,
        Dumper=KubernetesYamlDumper,
        default_flow_style=False,
        sort_keys=False,
        allow_unicode=True,
    )

load(yaml_str)

Load resources from a YAML string.

Parameters:

Name Type Description Default
yaml_str str

YAML string (may contain multiple documents separated by ---)

 required

Returns:

Type Description
list[dict]

List of resource dicts
Source code in src/k8smith/output/yaml.py 
89
90
91
92
93
94
95
96
97
98
def load(yaml_str: str) -> list[dict]:
    """Load resources from a YAML string.

    Args:
        yaml_str: YAML string (may contain multiple documents separated by ---)

    Returns:
        List of resource dicts
    """
    return [doc for doc in yaml.safe_load_all(yaml_str) if doc]