模块参考

pyrs_yaml 模块的完整 API 参考。

版本兼容

pyrs-yaml 以 ABI3 wheel 格式构建，单个 wheel 支持 Python 3.8 到 3.15，无需重新编译。

核心函数¶

`parse()`¶

将 YAML 字符串或字节解析为 YamlDocument。

parse(yaml: str | bytes, resolve_merges: bool = True, schema: str | dict = "core", max_depth: int = 1000, allow_duplicate_keys: bool = False) -> YamlDocument

参数:

yaml — str 或 bytes 的 YAML 内容
resolve_merges — 解析后是否解析合并键 (<<: *alias)（默认：True）
schema — Schema 名称 ("core", "json", "failsafe", "yaml1.1" 或已注册的自定义名称)，或内联 schema dict（参见 YAML Schema Language）
max_depth — 最大嵌套深度（默认：1000）
allow_duplicate_keys — 是否允许重复映射键（默认：False）

返回值: 包含解析后 YAML 的 YamlDocument

引发:

YamlParseError — 无效的 YAML 语法
YamlTypeError — 未找到指定的 Schema
TypeError — 输入不是 str 或 bytes

示例:

doc = pyrs_yaml.parse("key: value")
doc = pyrs_yaml.parse(b"key: value")
doc = pyrs_yaml.parse(yaml_str, schema="json")
doc = pyrs_yaml.parse(
    "addr: 0xFF", schema={"extends": "core", "rules": [{"pattern": "^0x[0-9a-fA-F]+$", "type": "int"}]}
)

`parse_file()`¶

解析 YAML 文件。

parse_file(path: str, schema: str | dict = "core", max_depth: int = 1000, allow_duplicate_keys: bool = False) -> YamlDocument

参数:

path — YAML 文件的路径
schema — Schema 名称或内联 dict（默认："core"）
max_depth — 最大嵌套深度（默认：1000）
allow_duplicate_keys — 是否允许重复映射键（默认：False）

返回值: YamlDocument

引发:

IOError — 文件未找到或无法读取
YamlParseError — 无效的 YAML

示例:

doc = pyrs_yaml.parse_file("config.yaml")

`parse_all_docs()`¶

从字符串解析多个 YAML 文档。

parse_all_docs(yaml: str) -> list[YamlDocument]

返回值: YamlDocument 对象列表

示例:

docs = pyrs_yaml.parse_all_docs("a: 1\n---\nb: 2")

PyYAML 兼容函数¶

`safe_load()`¶

解析 YAML 并返回原生 Python 类型。

safe_load(yaml: str) -> dict[str, Any] | list[Any]

等价于: PyYAML 的 yaml.safe_load()

示例:

data = pyrs_yaml.safe_load("key: value")  # {'key': 'value'}

`safe_loads()`¶

解析多个 YAML 文档。

safe_loads(yaml: str) -> list[dict[str, Any] | list[Any]]

等价于: PyYAML 的 yaml.safe_loads()

`safe_dump()`¶

将 Python 对象序列化为 YAML。

safe_dump(data: dict[str, Any] | list[Any] | ndarray) -> str

等价于: PyYAML 的 yaml.safe_dump()

支持的输入类型: dict, list, str, int, float, bool, None，以及 numpy.ndarray（所有维度和数值 dtype：int8/16/32/64, uint8/16/32/64, float32/64, complex64/128, bool）

`safe_dumps()`¶

safe_dump() 的别名。

safe_dumps(data: dict[str, Any] | list[Any] | ndarray) -> str

:material-json: 转换函数¶

`from_dict()`¶

将 Python dict 转换为 YAML 字符串。dict 的值中可以包含 numpy.ndarray。

from_dict(data: dict[str, Any]) -> str

`from_json()`¶

将 JSON 字符串转换为 YAML 字符串。

from_json(json_str: str) -> str

`dump_file()`¶

将 Python 对象序列化为 YAML 并写入文件。接受 dict、list 或 numpy.ndarray。

dump_file(data: Any, path: str) -> None

Pydantic 集成¶

`dump_pydantic()`¶

将 Pydantic 模型序列化为 YAML 字符串。

dump_pydantic(model: BaseModel) -> str

使用 model_dump(mode='json') 保持字符串类型（例如 "10001" 的邮政编码保持为字符串），然后委托给 safe_dump。

引发:

ImportError — 未安装 pydantic
TypeError — model 不是 Pydantic 的 BaseModel 实例

示例:

from pydantic import BaseModel
import pyrs_yaml


class User(BaseModel):
    name: str
    age: int


yaml_str = pyrs_yaml.dump_pydantic(User(name="Alice", age=30))

`parse_as()`¶

解析 YAML 字符串并针对 Pydantic 模型进行验证。

parse_as(model: type[BaseModel], src: str, **yaml_kwargs: Any) -> BaseModel

参数:

model — Pydantic 的 BaseModel 子类
src — 要解析的 YAML 字符串
**yaml_kwargs — 转发给 YAML() 构造函数的关键字参数

引发:

ImportError — 未安装 pydantic
TypeError — model 不是 Pydantic 的 BaseModel 子类
pydantic.ValidationError — 解析的数据未通过模型验证

示例:

user = pyrs_yaml.parse_as(User, "name: Alice\nage: 30")
print(user.name)  # Alice

`PyrsYamlConfigSettingsSource`¶

基于 pyrs-yaml 的 pydantic-settings YAML 配置来源类。pydantic_settings.YamlConfigSettingsSource 的即插即用替代品，使用 pyrs-yaml 而非 PyYAML 作为解析器。

PyrsYamlConfigSettingsSource(
    settings_cls: type[BaseSettings],
    yaml_file: ConfigFileSourceType | None = DEFAULT_PATH,
    yaml_file_encoding: str | None = None,
    yaml_config_section: str | None = None,
    deep_merge: bool = False,
)

从 SettingsConfigDict(yaml_file=...)（或直接传入）声明的 YAML 文件加载设置，并用 pyrs-yaml 解析器（YAML 1.2 核心 schema）解析。环境变量与 dotenv 覆盖、yaml_config_section（支持点号嵌套路径）、多文件 deep_merge、yaml_file_encoding 等其他 pydantic-settings 功能均与 YamlConfigSettingsSource 行为一致。

异常：

ImportError — 未安装 pydantic-settings 时抛出（安装 pyrs-yaml[settings]）

示例：

from pydantic_settings import BaseSettings, SettingsConfigDict
import pyrs_yaml


class Settings(BaseSettings):
    app_name: str

    model_config = SettingsConfigDict(yaml_file="config.yaml")

    @classmethod
    def settings_customise_sources(
        cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings
    ):
        return (
            init_settings,
            env_settings,
            dotenv_settings,
            file_secret_settings,
            pyrs_yaml.PyrsYamlConfigSettingsSource(settings_cls),
        )

注意： 该类采用惰性导出——import pyrs_yaml 不依赖 pydantic-settings。在未安装 pydantic-settings 时访问 pyrs_yaml.PyrsYamlConfigSettingsSource 会抛出带有安装提示的 ImportError。

标签注册表¶

`register_tag()`¶

注册自定义标签处理器。支持装饰器和命令式两种形式。

register_tag(name: str, handler: Callable | None = None, priority: int = 0) -> Callable

装饰器命令式

@pyrs_yaml.register_tag("!custom")
def handler(node):
    return f"custom:{node}"

pyrs_yaml.register_tag("!custom", handler_fn, priority=1)

`remove_tag()`¶

移除标签处理器。

remove_tag(name: str) -> None

`clear_tag_handlers()`¶

移除所有已注册的标签处理器。

clear_tag_handlers() -> None

YAML Schema Language¶

定义自定义 Schema，控制纯标量如何解析为 Python 类型。

`register_schema()`¶

注册一个自定义 Schema。

register_schema(name: str, schema: str | dict) -> None

参数:

name — Schema 名称
schema — YAML 字符串或 dict（包含 extends、rules、validate 键）

示例:

import pyrs_yaml

# 从 YAML 字符串注册自定义 Schema
pyrs_yaml.register_schema(
    "hex",
    """
name: hex
extends: core
rules:
  - pattern: ^0x[0-9a-fA-F]+$
    type: int
""",
)

# 使用自定义 Schema
y = pyrs_yaml.YAML(schema="hex")
doc = y.parse("addr: 0xFF")
assert doc.get("addr") == 255

d = pyrs_yaml.safe_load("addr: 0x1F", schema="hex")
assert d["addr"] == 31

内联 Schema dict¶

直接传入 dict 代替注册：

d = pyrs_yaml.safe_load(
    "addr: 0xFF",
    schema={
        "extends": "core",
        "rules": [{"pattern": "^0x[0-9a-fA-F]+$", "type": "int"}],
    },
)

extends — 可选的基 Schema（core、json、failsafe、yaml1.1）
rules — 有序的 {pattern, type} 列表；首个匹配生效
validate — 可选的结构校验规则：路径限定类型（$.port: int）、容器检查（sequence_of、mapping_of）和 required 存在性检查；使用 validate_against_schema(data, schema_yaml) 校验文档
支持的类型：null、bool、int、float、str
内置 Core Schema 仍使用零成本 match 分发（不受影响）
文件 I/O — load_schema(name, path) 从 YAML 文件加载 Schema；list_schemas() 返回所有已注册的 Schema

社区插件¶

定义自定义 YAML 节点类型，集成序列化和反序列化。

`CustomType`¶

自定义类型的基类。

class CustomType:
    python_type: type

    def from_yaml(self, value: str) -> Any: ...
    def to_yaml(self, obj: Any) -> str: ...
    def can_parse(self, node: CustomNode) -> bool: ...
    def validate(self, obj: Any) -> bool: ...

`register_type()`¶

注册自定义类型。

register_type(tag: str, type_handler: CustomType, priority: int = 0) -> None

示例:

from datetime import datetime


class TimestampType(pyrs_yaml.CustomType):
    python_type = datetime

    def from_yaml(self, value: str):
        return datetime.fromisoformat(value)

    def to_yaml(self, obj) -> str:
        return obj.isoformat()


pyrs_yaml.register_type("!timestamp", TimestampType())

# 加载：带标签的标量 → Python 对象
doc = pyrs_yaml.parse("when: !timestamp 2026-08-11T10:30:00")
assert isinstance(doc.get("when"), datetime)

# 导出：Python 对象 → 带标签的标量
data = {"ts": datetime(2026, 8, 11, 10, 30)}
out = pyrs_yaml.safe_dump(data)
# out 包含：ts: !timestamp 2026-08-11T10:30:00

方法	描述
`can_parse(node)`	该类型是否处理给定的 AST 节点
`from_yaml(value)`	将 YAML 字符串转换为 Python 对象
`to_yaml(obj)`	将 Python 对象转换为 YAML 字符串
`validate(obj)`	校验 Python 对象（返回 `bool`）

`remove_type()`¶

移除已注册的类型。

remove_type(name: str) -> None

`clear_type_handlers()`¶

移除所有已注册的类型处理器。

clear_type_handlers() -> None

合规性¶

`compliance_report()`¶

计算 YAML 测试套件的合规性报告。

compliance_report() -> dict

返回 YAML 测试套件的通过率及每个测试的结果。

流式事件¶

`parse_stream()`¶

增量解析 YAML，产出原始事件字典。

parse_stream(yaml: str) -> StreamIterator

返回一个 StreamIterator，每步产出一个事件字典。与 YAML().load_stream()（解析为 Python 值）不同，这暴露了原始令牌流。

`YamlStream`¶

YamlStream 类是一个惰性事件迭代器，由 YAML().load_stream() 和 YAML().load_stream_file() 返回。它逐个产出解析后的事件字典，无需将整个文档加载到内存中。

stream = yaml.load_stream_file("large.yaml")
for event in stream:
    print(event)

参见 YamlStream 了解完整的 API 详情。

异步函数¶

使用 asyncio.run_in_executor 的异步 I/O 包装器。在事件循环上下文中不阻塞。

`safe_dumps_async()`¶

将 Python 对象序列化为 YAML 字符串（异步）。

async def safe_dumps_async(data: Any) -> str

`safe_dump_async()`¶

将 Python 对象以 YAML 格式输出到 stdout（异步）。

async def safe_dump_async(data: Any) -> None

`safe_loads_async()`¶

将 YAML 字符串解析为原生 Python 对象（异步）。

async def safe_loads_async(yaml: str, schema: str = "core") -> Any

`safe_load_async()`¶

将 YAML 字符串解析为原生 Python 对象（异步）。

async def safe_load_async(yaml: str, schema: str = "core") -> Any

示例:

import asyncio, pyrs_yaml


async def main():
    yaml = await pyrs_yaml.safe_dumps_async({"a": 1})
    data = await pyrs_yaml.safe_loads_async(yaml)
    print(data)  # {'a': 1}


asyncio.run(main())

Markdown Front Matter¶

`read_markdown()`¶

从 Markdown 文件提取 YAML Front Matter。

read_markdown(path: str, schema: str = "core", max_depth: int = 1000) -> tuple[dict[str, Any] | None, str]

返回值: (frontmatter_dict, content_string)。没有Front Matter时，frontmatter 为 None。

`read_markdown_str()`¶

从 Markdown 字符串提取 YAML Front Matter。

read_markdown_str(content: str, schema: str = "core", max_depth: int = 1000) -> tuple[dict[str, Any] | None, str]

i18n 函数¶

`set_language()`¶

设置错误消息的语言。

set_language(lang: str) -> None

支持："en", "zh-CN", "ja-JP", "ko-KR"

`get_language()`¶

获取当前语言。

get_language() -> str

`list_languages()`¶

列出所有支持的语言。

list_languages() -> list[str]

`detect_language()`¶

从环境变量自动检测用户的首选语言。

detect_language() -> str

`negotiate_language()`¶

BCP 47 语言协商。

negotiate_language(user_locales: list[str], default: str = "en") -> str

异常¶

YamlParseError — YAML 解析错误（继承自 ValueError）
YamlSerializeError — YAML 序列化错误（继承自 ValueError）
YamlTypeError — 类型转换错误（继承自 TypeError）
YamlValidateError — JSON Schema 验证错误（继承自 ValueError）
YamlEditError — 就地编辑错误（继承自 ValueError）
YamlPathError — YAML 路径错误（继承自 ValueError）
YamlDocumentError — 过期的 Node 访问错误（继承自 Exception）

参见异常页面了解完整详情。

版本¶

__version__ = "0.14.0"

模块参考

核心函数¶

parse()¶

parse_file()¶

parse_all_docs()¶

PyYAML 兼容函数¶

safe_load()¶

safe_loads()¶

safe_dump()¶

safe_dumps()¶

:material-json: 转换函数¶

from_dict()¶

from_json()¶

dump_file()¶

Pydantic 集成¶

dump_pydantic()¶

parse_as()¶

PyrsYamlConfigSettingsSource¶

标签注册表¶

register_tag()¶

remove_tag()¶

clear_tag_handlers()¶

YAML Schema Language¶

register_schema()¶

内联 Schema dict¶

社区插件¶

CustomType¶

register_type()¶

remove_type()¶

clear_type_handlers()¶

合规性¶

compliance_report()¶

流式事件¶

parse_stream()¶

YamlStream¶

异步函数¶

safe_dumps_async()¶

safe_dump_async()¶

safe_loads_async()¶

safe_load_async()¶

Markdown Front Matter¶

read_markdown()¶

read_markdown_str()¶

i18n 函数¶

set_language()¶

get_language()¶

list_languages()¶

detect_language()¶

negotiate_language()¶

异常¶

版本¶

`parse()`¶

`parse_file()`¶

`parse_all_docs()`¶

`safe_load()`¶

`safe_loads()`¶

`safe_dump()`¶

`safe_dumps()`¶

`from_dict()`¶

`from_json()`¶

`dump_file()`¶

`dump_pydantic()`¶

`parse_as()`¶

`PyrsYamlConfigSettingsSource`¶

`register_tag()`¶

`remove_tag()`¶

`clear_tag_handlers()`¶

`register_schema()`¶

`CustomType`¶

`register_type()`¶

`remove_type()`¶

`clear_type_handlers()`¶

`compliance_report()`¶

`parse_stream()`¶

`YamlStream`¶

`safe_dumps_async()`¶

`safe_dump_async()`¶

`safe_loads_async()`¶

`safe_load_async()`¶

`read_markdown()`¶

`read_markdown_str()`¶

`set_language()`¶

`get_language()`¶

`list_languages()`¶

`detect_language()`¶

`negotiate_language()`¶