Edit on GitHub

sqlmesh.utils.yaml 

 1from __future__ import annotations
 2
 3import io
 4import typing as t
 5from os import getenv
 6from pathlib import Path
 7
 8from ruamel import yaml
 9
10from sqlmesh.utils.errors import SQLMeshError
11from sqlmesh.utils.jinja import ENVIRONMENT
12
13JINJA_METHODS = {
14    "env_var": lambda key, default=None: getenv(key, default),
15}
16
17YAML = lambda: yaml.YAML(typ="safe")  # noqa: E731
18
19
20def load(
21    source: str | Path,
22    raise_if_empty: bool = True,
23    render_jinja: bool = True,
24    allow_duplicate_keys: bool = False,
25) -> t.Dict:
26    """Loads a YAML object from either a raw string or a file."""
27    path: t.Optional[Path] = None
28
29    if isinstance(source, Path):
30        path = source
31        with open(source, "r", encoding="utf-8") as file:
32            source = file.read()
33
34    if render_jinja:
35        source = ENVIRONMENT.from_string(source).render(JINJA_METHODS)
36
37    yaml = YAML()
38    yaml.allow_duplicate_keys = allow_duplicate_keys
39    contents = yaml.load(source)
40    if contents is None:
41        if raise_if_empty:
42            error_path = f" '{path}'" if path else ""
43            raise SQLMeshError(f"YAML source{error_path} can't be empty.")
44        return {}
45
46    return contents
47
48
49@t.overload
50def dump(value: t.Any, stream: io.IOBase) -> None: ...
51
52
53@t.overload
54def dump(value: t.Any) -> str: ...
55
56
57def dump(value: t.Any, stream: t.Optional[io.IOBase] = None) -> t.Optional[str]:
58    """Dumps a ruamel.yaml loaded object and converts it into a string or writes it to a stream."""
59    result = io.StringIO()
60    yaml.YAML().dump(value, stream or result)
61
62    if stream:
63        return None
64    return result.getvalue()
def YAML(): 
18YAML = lambda: yaml.YAML(typ="safe")  # noqa: E731
def load( source: 'str | Path', raise_if_empty: bool = True, render_jinja: bool = True, allow_duplicate_keys: bool = False) -> Dict: 
21def load(
22    source: str | Path,
23    raise_if_empty: bool = True,
24    render_jinja: bool = True,
25    allow_duplicate_keys: bool = False,
26) -> t.Dict:
27    """Loads a YAML object from either a raw string or a file."""
28    path: t.Optional[Path] = None
29
30    if isinstance(source, Path):
31        path = source
32        with open(source, "r", encoding="utf-8") as file:
33            source = file.read()
34
35    if render_jinja:
36        source = ENVIRONMENT.from_string(source).render(JINJA_METHODS)
37
38    yaml = YAML()
39    yaml.allow_duplicate_keys = allow_duplicate_keys
40    contents = yaml.load(source)
41    if contents is None:
42        if raise_if_empty:
43            error_path = f" '{path}'" if path else ""
44            raise SQLMeshError(f"YAML source{error_path} can't be empty.")
45        return {}
46
47    return contents

Loads a YAML object from either a raw string or a file.

def dump( value: Any, stream: Union[io.IOBase, NoneType] = None) -> Union[str, NoneType]: 
58def dump(value: t.Any, stream: t.Optional[io.IOBase] = None) -> t.Optional[str]:
59    """Dumps a ruamel.yaml loaded object and converts it into a string or writes it to a stream."""
60    result = io.StringIO()
61    yaml.YAML().dump(value, stream or result)
62
63    if stream:
64        return None
65    return result.getvalue()

Dumps a ruamel.yaml loaded object and converts it into a string or writes it to a stream.