Python 库使用流程

rpkiparrot 的主要交付对象是可独立嵌入第三方应用的 Python 库。已有程序化数据集、验证、低层 RTR、在线自定义 JSON和事件及热配置可执行示例，真实入口见源码 API 参考。持久化、共享、诊断及远程示例见下文；当前未发布，完整验收状态以对应提交和报告为准。

离线读取和 ROV

此例使用完整的小型合成数据，显式选择离线参考时间。没有修改历史样本时间，也不把这份数据当作在线全网快照。

from datetime import datetime, timezone

from rpkiparrot import MemoryStore
from rpkiparrot.config import FreshnessPolicy
from rpkiparrot.models import EvaluationContext, OriginInput
from rpkiparrot.readers import parse_json
from rpkiparrot.validation import validate_origin, validate_origins

dataset = parse_json(
    {
        "metadata": {"generatedTime": "2026-10-01T00:00:00Z"},
        "roas": [{"asn": "AS64496", "prefix": "192.0.2.0/24", "maxLength": 24}],
        "aspas": [],
    },
    format="routinator-json",
)
store = MemoryStore(
    context=EvaluationContext.offline(at=datetime(2026, 10, 1, 0, 30, tzinfo=timezone.utc))
)
snapshot = store.replace_source("example-json", dataset, freshness=FreshnessPolicy(max_age=3600))
result = validate_origin(snapshot, "192.0.2.0/24", 64496, explain=True)
assert result.status.value == "valid"

batch = validate_origins(
    snapshot,
    [
        OriginInput(prefix="192.0.2.0/24", asn=64496, input_id="a"),
        OriginInput(prefix="192.0.2.0/25", asn=64496, input_id="b"),
        OriginInput(prefix="198.51.100.0/24", asn=64496, input_id="c"),
    ],
)
# 顺序保持 a/b/c；预期分别为 valid、invalid、notfound。
assert batch.complete

真实文件使用 read_json(path, format=...)。文件读取与解析同步，异步应用中的大文件交给受管理工作线程或 Client 的文件来源适配器。没有加载数据时不能调用验证并期待得到正常 notfound；空但完整有效的 roas 数组才具有这种语义。

标准 ASPA

AS_PATH 顺序为邻居在前、起源在后。此片段接续上面的快照：ASPA 已同步为空，所以多 AS 路径的授权数据不足会产生 unknown，区别于 ASPA 未加载异常。

from rpkiparrot.models import AsPath, AspaContext, NeighborRelationship
from rpkiparrot.validation import validate_aspa

aspa_result = validate_aspa(
    snapshot,
    AsPath.sequence((64497, 64496)),
    context=AspaContext(
        local_asn=64498,
        neighbor_asn=64497,
        relationship=NeighborRelationship.CUSTOMER,
    ),
    explain=True,
)
assert aspa_result.status.value == "unknown"

调用方负责 AS_PATH/AS4_PATH 重建和邻居上下文。BMP 观察缺少这些条件时调用独立 analyze_bmp_path，不补造邻居关系。完整规则见输入契约。

在应用任务组中运行 RTR

这个例子需要实际的 RTR 服务端点，文中的 example.net 地址仅为占位。anyio.run 属于示例应用入口，库内部不创建事件循环。

import anyio

from rpkiparrot import Client
from rpkiparrot.config import ClientConfig, RtrSourceConfig, SourceGroupConfig
from rpkiparrot.models import PayloadKind
from rpkiparrot.validation import validate_origin

config = ClientConfig(
    groups=(
        SourceGroupConfig(
            id="primary",
            priority=0,
            sources=(
                RtrSourceConfig(
                    id="cache-a",
                    host="cache.example.net",
                    transport="tls",
                    port=324,
                    trust_profile_id="operator-ca-v1",
                    server_name="cache.example.net",
                    ca_file="operator-ca.pem",
                    client_cert_file="router-cert.pem",
                    client_key_file="router-key.pem",
                ),
            ),
        ),
    ),
)


async def main() -> None:
    async with anyio.create_task_group() as tasks:
        async with Client(config) as client:
            await tasks.start(client.run)
            current = await client.wait_ready(required=frozenset({PayloadKind.VRP}), timeout=30.0)
            print(validate_origin(current, "192.0.2.0/24", 64496))
        # Client 先关闭，run 退出，然后应用任务组完成。


if __name__ == "__main__":
    anyio.run(main)

实际测试以本地协议替身运行同一生命周期，分别选择 asyncio 和 Trio 后端。连接成功不等于已获得数据；wait_ready 结束后保留的快照也可能随后到期，此时重新 get_snapshot。

上例用 ca_file 限定运营者 CA。省略它则使用 Python/OpenSSL 默认信任库，可能受 SSL_CERT_FILE / SSL_CERT_DIR 影响；也可以注入已加载客户端证书的 SSLContext。信任变化时同步修改 trust_profile_id；这个 ID 本身不会选择 CA。默认信任不免除双向 TLS、证书链和 SAN dNSName 验证。参见可执行传输示例和配置契约。

通过 SSH 连接 RTR 上游

安装 rpkiparrot[ssh] 后，将来源换为下列配置；其余 Client、任务组、就绪和查询流程相同。known_hosts 的主机密钥由运营者通过可信渠道分发，用户名/密钥需在服务端获准访问 rpki-rtr subsystem。示例域名和文件是部署占位。

from rpkiparrot.config import RtrSourceConfig, SshConfig

source = RtrSourceConfig(
    id="ssh-cache",
    host="cache.example.net",
    transport="ssh",  # 默认 22，可显式指定服务端端口
    trust_profile_id="operator-ssh-v1",
    ssh=SshConfig(
        username="rtr-reader",
        known_hosts="credentials/known_hosts",
        client_keys=("credentials/id_ed25519",),
    ),
)

加密密钥可传 passphrase_provider，密码认证可传 password_provider；均为准备阶段调用的同步回调。运行应用使用 asyncio；内置适配器不支持 Trio。新凭据通过来源替换与新 trust_profile_id 提交。无需真实上游的可执行 SSH 示例生成临时密钥并演示本地认证、完整同步和 ROV；它的服务端只作为示例夹具，不是库提供的 RTR 服务。

观察变化和重新同步

以下协程由宿主在已经运行的 Client 上调用。为使“跳到更新的当前快照”仍然正确，例子每次重新验证应用持有的全部路由；大型应用可改用事件给出的保守影响范围，但必须维护自己的路由索引和事件连续性。

from rpkiparrot.errors import ResyncRequiredError
from rpkiparrot.models import OriginInput, SnapshotEvent
from rpkiparrot.validation import validate_origins

routes = (OriginInput(prefix="192.0.2.0/24", asn=64496, input_id="route-1"),)


def revalidate_all(snapshot) -> None:
    results = validate_origins(snapshot, routes)
    for item in results.items:
        print(item.input_id, item.result, item.error)


async def observe(client) -> None:
    while True:
        try:
            async with client.watch() as subscription:
                revalidate_all(subscription.initial)
                async for event in subscription:
                    if isinstance(event, SnapshotEvent):
                        revalidate_all(await client.get_snapshot())
            return
        except ResyncRequiredError:
            # 重建订阅时同时取得新 initial，不分开读取和注册。
            continue

正常 Client 关闭使循环结束，取消向宿主传播。批量的逐条不可用也通过 item.error 检查；不要只检查函数有没有抛异常。处理慢时不得静默跳过 ResyncRequired。

运行中增删来源和修改配置

下面的设计示例由宿主在已运行的 Client 上调用，current 是宿主保存的当前冻结配置。每次传入取得配置时对应的 revision；其他任务先修改了配置时应处理 ConfigurationConflictError，重新构造候选，不能盲目覆盖。

from dataclasses import replace

from rpkiparrot.config import ClientConfig, RtrSourceConfig


async def add_cache(client, current: ClientConfig, revision):
    primary = current.groups[0]
    new_source = RtrSourceConfig(
        id="cache-b",
        host="cache-b.example.net",
        port=323,
        transport="tcp",
        trust_profile_id="operator-cache-b-v1",
    )
    candidate = replace(
        current,
        groups=(replace(primary, sources=primary.sources + (new_source,)),) + current.groups[1:],
    )
    receipt = await client.apply_config(candidate, expected_revision=revision)
    return candidate, receipt


async def change_query_timeout(client, current: ClientConfig, revision):
    primary = current.groups[0]
    source = primary.sources[0]  # 此例要求它是 RtrSourceConfig。
    candidate = replace(
        current,
        groups=(
            replace(primary, sources=(replace(source, query_timeout=180),) + primary.sources[1:]),
        )
        + current.groups[1:],
    )
    receipt = await client.apply_config(candidate, expected_revision=revision)
    return candidate, receipt


async def remove_all_sources(client, current: ClientConfig, revision):
    candidate = replace(current, groups=())
    receipt = await client.apply_config(candidate, expected_revision=revision)
    return candidate, receipt

增加来源的回执不代表该来源已同步；按来源状态和事件观察结果。删除单个来源时按 ID 过滤对应 sources，再以同一 apply_config 提交；不要就地修改元组、SSLContext 或注入适配器。删除后 Client 可继续接受新增来源。新配置不回写已经取得的快照，撤换信任来源后应获取新快照并重验，见完整变更契约。

在线自定义 JSON 格式与 ASPA 遍历

宿主显式创建 reader 并绑定来源，不需要更改模块级注册表。此例的 adapter 由宿主实现 JsonFormatAdapter，source 的 format 与 reader_profile_id 必须明确且与它相符。

from rpkiparrot import Client
from rpkiparrot.config import ClientConfig, SourceGroupConfig
from rpkiparrot.query import iter_aspas
from rpkiparrot.readers import JsonReader


def client_for_json(source, adapter) -> Client:
    config = ClientConfig(groups=(SourceGroupConfig(id="json", priority=0, sources=(source,)),))
    return Client(config, readers={source.id: JsonReader(adapters=(adapter,))})


def list_customers(snapshot) -> None:
    for entry in iter_aspas(snapshot):
        print(entry.customer, entry.providers)

Client 按前面的任务组示例打开和运行。ASPA 已同步为空时迭代零项，未加载或到期时明确报错；迭代期间固定快照，不逐项偷偷取得新版本。需要单一来源的原始集合时使用来源诊断/完整导出，source_id 筛选后的生效遍历仍返回合并 provider 集合。

可选能力

需求	使用方式
SQLite/DuckDB 恢复	显式 PersistenceConfig；需要确认落盘时 await client.flush 并检查回执
本机多个进程读取	SQLite SharedSnapshotClient；使用自身快照身份并保留 upstream_id
远程服务	http extra 中的 RemoteClient；异步返回相同领域结果，使用 SnapshotRef 固定远程版本
自定义连接	按来源 ID 注入 TransportFactory，每次 connect 交付新 ByteStream
只解码 RTR	独立 PduDecoder，不创建 Client 或打开网络
只加载导出文件	load_snapshot，保留原始来源与期限，不重新设置 max_age

分发包含核心 rpkiparrot 和按需 extras，例如 rpkiparrot[cli]、rpkiparrot[sqlite]、rpkiparrot[duckdb]、rpkiparrot[http]、rpkiparrot[service]、rpkiparrot[trio]、rpkiparrot[ssh]。当前工作区可通过 uv sync --locked --all-extras --group dev 安装开发环境；正式上传和发布版本另行公布。

显式验证指标

核心包提供 rpkiparrot.metrics.MetricsCollector。使用 collector.validate_origin(snapshot, prefix, asn) 或相应批量/ASPA/BMP 方法代替需要计数的直接调用，再以 collector.get_metrics() 采样。输入、返回类型、异常和固定快照时效规则与纯函数相同；直接调用纯函数不会改动该收集器。默认不计时，MetricsCollector(timing=True) 同时累计耗时。每个收集器独立，无后台任务，可由受管理线程共享。

可执行指标示例验证混合批次的调用数、条目数与错误数。正常的 invalid 或 unknown 结果不表示执行错误；字段单位和计数边界见状态与指标。Client.get_metrics 另外提供来源、快照、事件与持久化指标，HTTP 服务主动收集其自身的领域验证调用。

ASPA 地址族与持久化示例

按地址族验证运行独立的 IPv4/IPv6 授权及验证。给查询和验证传 afi=Afi.IPV4 或 afi=Afi.IPV6；不指定 AFI 只适用于两族完整视图等价的情况。Client.wait_ready(..., aspa_afi=Afi.IPV4) 可等待单族，省略时等待两族分别可用。

在线持久化与恢复可在 SQLite/DuckDB 及 asyncio/Trio 组合运行，实际完成来源同步、flush、关闭和原期限内恢复。恢复发生在 Client 上下文进入时，开始 run 之前即可查询；未启动 run 的只读恢复检查不提交新的 head。CLI 持久化用独立子进程验证写入与 SQLite 共享读取。固定线程后端的同步使用见后端示例，共享轮询生命周期见共享示例。

远程 HTTP 调用

安装 rpkiparrot[http] 后，使用 async with RemoteClient(base_url) 管理 HTTP 连接。构造不联网，导入不要求服务框架。所有网络方法需要 await，领域结果仍是 OriginResult、AspaResult 等核心数据类。get_snapshot() 返回固定的 SnapshotRef；将其传给验证或遍历的 snapshot= 参数，可固定同一版本。ASPA 的 afi 与嵌入式接口相同，不默认合并两族。

iter_vrps 和 iter_aspas 是异步生成器，page_size 默认使用服务端配置，分页全程保留同一 token 和过滤条件；提前结束时使用 contextlib.aclosing。到期或淘汰明确失败，不以新快照补齐余页。watch() 是异步上下文，返回原子取得的 initial 和可异步遍历的后续事件；resync_required 要求重新观察，watch_conflict 可沿原游标重试。响应体和总请求时长均有界，线程完成解析后才释放所占资源。

可执行 SDK 示例默认使用独立固定 HTTP 响应，无需外部服务器，并可选择 asyncio/Trio；显式 --url 改为实际服务。这个 mock 示例不作为真实服务端的互操作或在线信任证据。完整端点、认证和生命周期见 HTTP 契约。

RTR 录制与离线回放

显式 Recorder 包装一个 RTR 会话的传输，记录应用层字节；会话参数和协议资源限制必须与 recorder header 一致。先关闭会话，再关闭 recorder，才能得到完整连接轨迹。录制超限或写入失败不阻塞来源同步，但报告不完整。replay(path, at=...) 同步离线检查录制完整性及协议状态机差异，不启动网络或发布在线数据。

可执行录制和回放示例在宿主管理的 asyncio/Trio 上执行；它只证明所示受控会话。matched 不代表实际宿主接纳或持久化已验证，完整边界见诊断契约。