fromyaml 上下文方法可以用来加载yaml 内容,属于一个工具类,比如automate-dv 就使用了不少方法
参考使用
{%- set info -%}
source_model:
raw_staging: "raw_customer"
derived_columns:
SOURCE: "!1"
LOAD_DATETIME: "CRM_DATA_INGESTION_TIME"
EFFECTIVE_FROM: "BOOKING_DATE"
START_DATE: "BOOKING_DATE"
END_DATE: "TO_DATE('9999-12-31')"
hashed_columns:
CUSTOMER_HK: "CUSTOMER_ID"
NATION_HK: "NATION_ID"
CUSTOMER_NATION_HK:
- "CUSTOMER_ID"
- "NATION_ID"
CUSTOMER_HASHDIFF:
is_hashdiff: true
columns:
- "CUSTOMER_NAME"
- "CUSTOMER_ID"
- "CUSTOMER_PHONE"
- "CUSTOMER_DOB"
{%- endset -%}
{% set metadata_dict = fromyaml(info) %}
{{log(metadata_dict, info=True)}}
内部处理
- 实际上就是包装了一个context 方法
@contextmember()
@staticmethod
def fromyaml(value: str, default: Any = None) -> Any:
"""The fromyaml context method can be used to deserialize a yaml string
into a Python object primitive, eg. a `dict` or `list`.
:param value: The yaml string to deserialize
:param default: A default value to return if the `string` argument
cannot be deserialized (optional)
Usage:
{% set my_yml_str -%}
dogs:
- good
- bad
{%- endset %}
{% set my_dict = fromyaml(my_yml_str) %}
{% do log(my_dict['dogs'], info=true) %}
-- ["good", "bad"]
{% do my_dict['dogs'].pop() }
{% do log(my_dict['dogs'], info=true) %}
-- ["good"]
"""
try:
return safe_load(value)
except (AttributeError, ValueError, yaml.YAMLError):
return default
- safe_load 方法
def safe_load(contents) -> Optional[Dict[str, Any]]:
return yaml.load(contents, Loader=SafeLoader)
参考资料
https://docs.getdbt.com/reference/dbt-jinja-functions/fromyaml
https://automate-dv.readthedocs.io/en/latest/