YAML DSL 用户指导文档¶

name: minimal_order_report

main_source:
  source_id: orders
  loader: "myapp.loaders:load_orders"
  fields:
    order_id:
      name: 订单ID

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunOutputOptions, DemandRunSecurityOptions, RunOverrides, run

result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        outputs=DemandRunOutputOptions(
            overrides=RunOverrides.csv_file(
                output_root="./output",
                fields=["order_id"],
                file_id="minimal_order_report",
            ),
        ),
    ),
)

resources:
  books:
    minimal_order_report:
      xlsx:
        path: ./output

outputs:
  - name: detail
    to: {book: minimal_order_report, sheet: 明细}
    fields: [order_id]

name: simple_order_report
description: 简单订单报表

main_source:
  source_id: orders
  loader: "myapp.loaders:load_orders"
  params:
    begin_time: "2024-01-01"
    end_time: "2024-01-31"
  fields:
    order_id:
      name: 订单ID
    amount:
      name: 订单金额

sources:
  customers:
    loader: "myapp.loaders:load_customers"
    key: customer_id
    params:
      customer_ids: {$keys: {as: set}}
    fields:
      customer_name:
        name: 客户名称

relations:
  orders_to_customers:
    steps:
      - from: orders.customer_id
        to: customers.customer_id

fields:
  total_amount:
    name: 总金额
    compute: "sum(amount)"

outputs:
  - name: detail
    to: {file: order_report}
    write: {header_fields_output_by: name}
    fields: [order_id, customer_name, amount, total_amount]

resources:
  files:
    order_report:
      csv_file:
        path: ./output

# 安装独立 CLI 包(会拉取 `scalim`)
uv tool install scalim-cli

# 验证配置语法
scalim-cli yaml-dsl validate config.yaml

from scalim.dsl.yaml_dsl import (
    DemandRunOptions,
    DemandRunOutputOptions,
    DemandRunSecurityOptions,
    RunOverrides,
    WorkflowRunOptions,
    run,
    run_workflow,
)

# 加载并执行 YAML 配置(安全:需要 allowlist)
result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
    ),
)

# workflow YAML 执行入口
workflow_result = run_workflow(
    "path/to/workflow.yaml",
    options=WorkflowRunOptions(
        demand=DemandRunOptions(security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"]))),
    ),
)

# 常见用法: YAML 不声明 outputs,由 Python 调用侧 RunOverrides 指定单输出编排(推荐)
result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        outputs=DemandRunOutputOptions(
            overrides=RunOverrides.csv_file(
                output_root="./output",
                fields=["order_id"],
                file_id="order_report",
            ),
        ),
    ),
)

main_source:
  params:
    end_dt: {$init_var: end_dt}

resources:
  files:
    detail_csv:
      csv_file:
        path: {$init_var: out_root}

outputs:
  - name: detail
    to: {file: detail_csv}
    fields: [order_id]

resources:
  books:
    report:
      xlsx:
        path: {$init_var: out_root}

from datetime import datetime
from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunSecurityOptions, DemandRunTemplateOptions, run

result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        template=DemandRunTemplateOptions(
            init_vars={
                "end_dt": datetime(2024, 1, 31),
                "out_root": "./output",
            },
        ),
    ),
)

main_source:
  fields:
    order_id:
      extract: order_id    # 可省略: 缺省等价于 extract: <field_id>
      name: 订单ID          # 显示名称
      value_cast: int      # 值转换(可选)

fields:
  profit:
    name: 利润
    compute: "amount - cost"  # Python 表达式(使用 field_id 作为变量;依赖从表达式推导)

relations:
  orders_to_customers:
    steps:
      - from: orders.customer_id    # 上游字段
        to: customers.customer_id   # 下游字段

resources:
  books:
    report:
      xlsx:
        path: ./output

_templates:
  report_to: &report_to {book: report}

outputs:
  - name: detail
    to: {<<: *report_to, sheet: 明细}
    fields: [order_id, customer_name, amount]

  - name: direct
    from: detail
    to: {<<: *report_to, sheet: 直客}
    where: "channel == 'direct'"

  - name: by_channel
    to: {<<: *report_to, sheet: 渠道汇总}
    aggregate:
      group_by: [channel]
      fields:
        order_cnt: {count: {}}
        sum_amount: {sum: {field: amount}}

name: order_report
description: 订单报表配置

_templates:
  # 定义可复用的 YAML 锚点
  common_params_keys: &common_params_keys
    ids: {$keys: {as: set}}

main_source:
  source_id: orders
  loader: "myapp.loaders:load_orders"
  params:
    begin_time: "2024-01-01"
    end_time: "2024-01-31"
    status: "completed"
  fields:
    order_id:
      name: 订单ID
    customer_id:
      name: 客户ID
    amount:
      name: 订单金额
      value_cast: int

sources:
  customers:
    loader: "myapp.loaders:load_customers"
    key: customer_id
    params:
      customer_ids_set: {$keys: {as: set}}

sources:
  customers:
    loader: "myapp.loaders:load_customers_by_rows"
    key: customer_id
    params:
      rows: {$rows: {cache_mode: batch}}

sources:
  regions:
    loader: "myapp.loaders:load_regions"
    key: region_id
    lookup_cast:
      int: {}                  # 转换类型:auto/int/str/sep_first

# 处理 "1,2,3" 这样的 CSV 多值字段
lookup_cast:
  sep_first:
    sep: ","        # 默认 ","

sources:
  payment_methods:
    loader: "myapp.loaders:load_payment_methods"
    key: payment_method_id
    normalize:
      index_by_key:
        on_conflict: error

normalize:
  take_first:
    on_empty: miss  # miss|null|error

normalize:
  project_fields:
    on_missing: error  # error|null
    fields:
      order_id: {from_key: true}
      customer_level: {extract: "[1].clearn_reason_level"}
      operation_level: {extract: "[2].clearn_reason_level"}
      review_status: {extract: review_status}

normalize:
  map_values:
    steps:
      - take_first:
          on_empty: miss
      - project_fields:
          on_missing: error
          fields:
            order_id: {from_key: true}
            review_status: {extract: review_status}

main_source:
  fields:
    # 顶层同名 key: extract 可省略
    review_status:
      name: 审核状态

    # int-key nested dict: role_id 是 int(例如 1/2)
    customer_clearn_reason_level:
      name: 客户净利原因等级
      extract: "[1].clearn_reason_level"

    # dotted literal key: row 的 key 就叫 "a.b"
    dotted_literal_key_value:
      extract: '["a.b"].x'

result = {
  1: {1: {"clearn_reason_level": 2}, 2: {"clearn_reason_level": 1}, "review_status": 0},
}

sources:
  clearn_reasons:
    fields:
      customer_level:
        extract: "[1].clearn_reason_level"
      operation_level:
        extract: "[2].clearn_reason_level"
      review_status:
        extract: review_status

{"customer_level": 2, "operation_level": 1, "review_status": 0}

main_source:
  fields:
    order_id:
      name: 订单ID
    customer_id:
      name: 客户ID

sources:
  customers:
    fields:
      customer_name:
        name: 客户名称
        relation: *orders_to_customers  # YAML 别名引用
      customer_level:
        name: 会员等级
        value_cast: str
        relation:
          steps:               # 内联关系定义
            - from: orders.customer_id
              to: customers.customer_id

fields:
  profit:
    name: 利润
    compute: "amount - cost"

  status_text:
    name: 状态文本
    call_by: "myapp.enums:get_status_text(status=status, ctx=$ctx)"

  profit_margin:
    name: 利润率
    compute: "(amount - cost) / amount * 100 if amount > 0 else 0"

  final_price:
    name: 最终价格
    compute: "order_amount * price_adjustment + shipping_fee"

relations:
  orders_to_customers:
    steps:
      - from: orders.customer_id
        to: customers.customer_id

  orders_to_categories:  # 多级关联
    steps:
      - from: orders.product_id
        to: products.product_id
      - from: products.category_id
        to: categories.category_id

  orders_to_region_pricing:  # 复合键关联
    steps:
      - from: [orders.region_id, orders.product_category_id]
        to: [region_pricing.region_id, region_pricing.product_category_id]

sources:
  customers:
    fields:
      customer_name:
        name: 客户名称
        relation:
          steps:
            - from: orders.customer_id
              to: customers.customer_id

main_source:
  source_id: orders
  loader: "myapp.loaders:load_orders"
  fields:
    customer_id:
      extract: customer_id_col  # data_key

sources:
  customers:
    loader: "myapp.loaders:load_customers"
    key: customer_id
    params:
      ids: {$keys: {as: set}}
    fields:
      customer_id:
        extract: id  # data_key

relations:
  orders_to_customers:
    steps:
      - from: orders.customer_id     # field_id
        to: customers.customer_id    # field_id

sources:
  customers:
    loader: "myapp.loaders:load_customers"
    key: customer_id
    params:
      ids: {$keys: {as: list}}

steps:
  - from: orders.customer_id
    to: customers.customer_id
    lookup_cast:
      int: {}

relations:
  orders_to_customers:
    steps:
      - from: orders.customer_id
        to: customers.customer_id

relations:
  orders_to_countries:
    steps:
      - from: orders.pay_id
        to: pays.pay_id
      - from: pays.country_id
        to: countries.country_id

relations:
  orders_to_region_pricing:
    steps:
      - from: [orders.region_id, orders.product_category_id]
        to: [region_pricing.region_id, region_pricing.product_category_id]

relations:
  orders_to_small_groups:
    steps:
      - from: orders.small_group_ids
        to: small_groups.small_group_id
        lookup_cast:
          sep_first:
            sep: ","

outputs:
  - name: detail
    to: {file: detail_csv}
    write: {header_fields_output_by: name}
    fields: [order_id, customer_name, amount, profit]

resources:
  files:
    detail_csv:
      csv_file:
        path: ./output

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.logs import LoggingObserver, PrettyLoggingObserver

run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[PrettyLoggingObserver()]),  # 或 LoggingObserver()
    ),
)

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.performance import PerformanceConfig, PerformanceObserver, PerformanceThresholds

perf = PerformanceObserver(
    PerformanceConfig(
        metrics={"duration", "memory", "cpu"},
        sampling_interval=2,
        report_format="csv",
        output_path="./output/perf_report.csv",
        include_details=True,
        thresholds=PerformanceThresholds(batch_duration_warn=1.5, memory_increase_warn=256),
    )
)

run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[perf]),
    ),
)

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.relations import RelationConfig, RelationObserver

relations = RelationObserver(
    RelationConfig(
        sampling_rate=0.05,
        log_type_mismatch=True,
        report_format="json",
        output_path="./output/relations_report.json",
        include_details=True,
    )
)

run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[relations]),
    ),
)

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunOutputOptions, DemandRunSecurityOptions, RunOverrides, run
from scalim.ob.presets.viz import VizObserverConfig

run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        outputs=DemandRunOutputOptions(
            overrides=RunOverrides(
                viz_config=VizObserverConfig(
                    output_dir="./output",
                    trace_enabled=False,
                    payload_policy="summary",
                    run_name="order_report",
                    env="production",
                )
            ),
        ),
    ),
)

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.execution_trace import ExecutionTraceObserver

trace = ExecutionTraceObserver()
_ = run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[trace]),
    ),
)

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.row_gap import RowGapObserver

row_gap = RowGapObserver(primary_loader_name="tickets", data_loader_names=["customers", "agents"], sample_limit=3)
_ = run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[row_gap]),
    ),
)

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.memory import MemoryOptimizationObserver

mem = MemoryOptimizationObserver(auto_report=True, max_fields=20)
_ = run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[mem]),
    ),
)

_templates:
  # 定义可复用的配置
  common_field: &common_field
    name: 默认名称
    value_cast: auto

  step_to_customers: &step_to_customers
    from: orders.customer_id
    to: customers.customer_id

  params_keys: &params_keys
    ids: {$keys: {as: set}}

relations:
  orders_to_customers:
    steps:
      - *step_to_customers

sources:
  customers:
    params: *params_keys

  products:
    params: *params_keys

_templates:
  base_step: &base_step
    from: orders.customer_id
    to: customers.customer_id

relations:
  orders_to_customers_default:
    steps:
      - *base_step

  orders_to_customers_custom:
    steps:
      - <<: *base_step            # 继承 base_step
        lookup_cast:              # 添加自定义属性
          name: int

sources:
  customers:
    loader: "myapp.loaders:load_customers"
    key: customer_id
    params:
      customer_ids_set: {$keys: {as: set}}

def load_customers(customer_ids_set: set[int]) -> dict:
    pass

sources:
  customers:
    loader: "myapp.loaders:load_customers_by_rows"
    key: customer_id
    params:
      rows: {$rows: {cache_mode: batch}}

def load_customers_by_rows(rows: list[dict]) -> dict:
    pass

lookup_cast:
  auto: {}          # 自动推断类型并归一化

lookup_cast:
  int: {}           # 转为整数

lookup_cast:
  str: {}           # 转为字符串

# 假设 orders.small_group_ids = "1,2,3"
relations:
  orders_to_small_groups:
    steps:
      - from: orders.small_group_ids
        to: small_groups.small_group_id
        lookup_cast:
          sep_first:
            sep: ","      # 按逗号分割,取 "1"

sources:
  customers:
    cache_mode: none        # 每次关联都重新查询
    params:
      ids: {$keys: {as: set}}

sources:
  order_types:
    cache_mode: preload_forever    # 启动时加载一次,永久缓存
    params:
      group_by: type_id

from scalim.dsl.yaml_dsl import DemandRunRuntimeOptions, LookupChunking

runtime = DemandRunRuntimeOptions(
    lookup_chunking={
        "customers": LookupChunking.sized(800),                   # 串行分片
        # "customers": LookupChunking.sized(800, parallel=True), # 片间并行（须 parallel_mode=adaptive）
    },
    parallel_mode="adaptive",  # 仅当 sized(..., parallel=True) 时需要
)

resources:
  books:
    report:
      xlsx:
        path: ./output

_templates:
  report_to: &report_to {book: report}

outputs:
  - name: detail
    to: {<<: *report_to, sheet: 明细}
    fields: [order_id, order_source, amount, cost, profit]
  - name: summary
    to: {<<: *report_to, sheet: 汇总}
    aggregate:
      group_by: [order_source]
      fields:
        order_cnt: {count: {field: order_id}}
        sum_amount: {sum: {field: amount}}
        sum_profit: {sum: {field: profit}}

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunOutputOptions, DemandRunSecurityOptions, OutputExtrasOverride, RunOverrides, run

result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        outputs=DemandRunOutputOptions(overrides=RunOverrides(output_extras=OutputExtrasOverride(meta=True, audit=True))),
    ),
)

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunOutputOptions, DemandRunSecurityOptions, RunOverrides, run

result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        outputs=DemandRunOutputOptions(
            overrides=RunOverrides.xlsx_file_single_sheet(
                output_root="./output",
                fields=["order_id", "amount", "profit"],
                sheet="明细",
                book_id="report",
            ),
        ),
    ),
)

from scalim.dsl.yaml_dsl.tools import derive_base_module_path, load_output_config

cfg = load_output_config("path/to/config.yaml")
base_module_path = derive_base_module_path("path/to/config.yaml")

name: order_report
description: 订单报表 - 展示客户、支付、国家等关联信息

main_source:
  source_id: orders
  loader: "scalim_misc.example_report_ir:DAL.paged_get_order_list"
  params:
    begin_time: "2024-01-01"
    end_time: "2024-01-07"
  fields:
    order_id:
      name: 订单ID

    amount:
      name: 金额

    cost:
      name: 成本

    customer_id:
      name: 客户ID

    pay_id:
      name: 支付ID

relations:
  orders_to_customers:
    steps:
      - from: orders.customer_id
        to: customers.customer_id

  orders_to_pays:
    steps:
      - from: orders.pay_id
        to: pays.pay_id

sources:
  customers:
    loader: "scalim_misc.example_report_ir:BLL.get_customer_info_from_api_of_kw_params"
    key: customer_id
    params:
      customer_ids_set: {$keys: {as: set}}
    fields:
      customer_name:
        name: 客户名称
        relation: *orders_to_customers

  pays:
    loader: "scalim_misc.example_report_ir:BLL.get_pay_info_from_api_of_concrete_params"
    key: pay_id
    params:
      pay_ids_set: {$keys: {as: set}}
    fields:
      pay_method:
        name: 支付方式
        relation: *orders_to_pays

fields:
  profit:
    name: 利润
    compute: "amount - cost"

  profit_margin:
    name: 利润率
    compute: "(amount - cost) / amount * 100 if amount > 0 else 0"

outputs:
  - name: detail
    to: {file: detail_csv}
    fields: [order_id, customer_name, amount, cost, profit]

resources:
  files:
    detail_csv:
      csv_file:
        path: ./.tmp/output

name: ecommerce_order_report
description: 电商订单报表

_templates:
  step_orders_to_customers: &step_orders_to_customers
    from: orders.customer_id
    to: customers.customer_id

  step_orders_to_products: &step_orders_to_products
    from: orders.product_id
    to: products.product_id

main_source:
  source_id: orders
  loader: "scalim_misc.demo_big_data_report.loaders:load_orders"
  params:
    field_keys:
      - order_id
      - order_date
      - quantity
      - unit_price
      - discount_rate
      - customer_id
      - product_id
  fields:
    order_id:
      name: 订单ID

    quantity:
      name: 数量
      value_cast: int

    unit_price:
      name: 单价

    discount_rate:
      name: 折扣率

relations:
  orders_to_customers:
    steps:
      - *step_orders_to_customers

  orders_to_categories:  # 多级关联
    steps:
      - *step_orders_to_products
      - from: products.category_id
        to: categories.category_id

  orders_to_region_pricing:  # 复合键关联
    steps:
      - from: [orders.region_id, orders.product_category_id]
        to: [region_pricing.region_id, region_pricing.product_category_id]

sources:
  customers:
    loader: "scalim_misc.demo_big_data_report.loaders:load_customers"
    key: customer_id
    params:
      ids: {$keys: {as: set}}
    fields:
      customer_name:
        name: 客户姓名
        relation: *orders_to_customers

      customer_level:
        name: 会员等级
        relation: *orders_to_customers

  products:
    loader: "scalim_misc.demo_big_data_report.loaders:load_products"
    key: product_id
    lookup_cast:
      int: {}
    params:
      ids: {$keys: {as: list}}
    fields:
      product_name:
        name: 产品名称
        relation: *orders_to_products

      product_category_id:
        name: 产品分类ID
        extract: category_id
        relation: *orders_to_products

  categories:
    loader: "scalim_misc.demo_big_data_report.loaders:load_categories"
    key: category_id
    params:
      ids: {$keys: {as: set}}
    fields:
      category_name:
        name: 产品分类
        relation: *orders_to_categories

  region_pricing:
    loader: "scalim_misc.demo_big_data_report.loaders:load_region_pricing"
    key: [region_id, product_category_id]
    cache_mode: preload_forever
    fields:
      price_adjustment:
        name: 价格调整系数
        relation: *orders_to_region_pricing

      shipping_fee:
        name: 运费
        relation: *orders_to_region_pricing

      tax_rate:
        name: 税率
        relation: *orders_to_region_pricing

fields:
  order_amount:
    name: 订单金额
    compute: "quantity * unit_price * discount_rate"

  profit:
    name: 利润
    compute: "order_amount - product_cost * quantity"

  tax_amount:
    name: 税费
    compute: "order_amount * tax_rate"

  final_price:
    name: 最终价格
    compute: "order_amount * price_adjustment + shipping_fee"

outputs:
  - name: detail
    to: {file: detail_csv}
    write: {header_fields_output_by: name}
    fields:
      - order_id
      - customer_name
      - quantity
      - unit_price
      - product_name
      - category_name
      - price_adjustment
      - shipping_fee
      - tax_rate
      - order_amount
      - profit
      - tax_amount
      - final_price

resources:
  files:
    detail_csv:
      csv_file:
        path: ./.tmp/output

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.performance import PerformanceConfig, PerformanceObserver
from scalim.ob.presets.relations import RelationConfig, RelationObserver

_ = run(
    "path/to/ecommerce_report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["scalim_misc.demo_big_data_report.loaders"])),
        runtime=DemandRunRuntimeOptions(
            components=[
                PerformanceObserver(PerformanceConfig(metrics={"duration", "memory", "cpu"})),
                RelationObserver(RelationConfig(sampling_rate=0.05, report_format="json", output_path="./output/relations_report.json")),
            ],
        ),
    ),
)

# 推荐
source_id: order_details
field_id: customer_name

# 不推荐
source_id: OrderDetails
field_id: customerName

_templates:
  # 关系步骤模板
  steps:
    to_customer: &step_to_customer
      from: orders.customer_id
      to: customers.customer_id

    to_product: &step_to_product
      from: orders.product_id
      to: products.product_id

  # params 模板片段
  params:
    keys_set: &params_keys_set
      ids: {$keys: {as: set}}

    keys_list: &params_keys_list
      ids: {$keys: {as: list}}

  # 字段配置模板
  fields:
    customer_name: &field_customer_name
      name: 客户姓名
      relation: *step_to_customer

# 使用模板
relations:
  orders_to_customers:
    steps:
      - *step_to_customer

sources:
  customers:
    params: *params_keys_set
    fields:
      customer_name: *field_customer_name

  products:
    params: *params_keys_list

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run

# 小数据集或低内存环境
run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp"])),
        runtime=DemandRunRuntimeOptions(batch_size=500),
    ),
)

# 大数据集或高内存环境
run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp"])),
        runtime=DemandRunRuntimeOptions(batch_size=2000),
    ),
)

sources:
  # 维度表(数据量小且不变)使用 preload_forever
  order_types:
    cache_mode: preload_forever

  # 事实表(数据量大)使用 none
  orders:
    cache_mode: none

resources:
  files:
    detail_csv:
      csv_file:
        path: ./output

outputs:
  - name: detail
    to: {file: detail_csv}
    fields: [order_id]

# 推荐:`$keys`
sources:
  customers:
    params:
      ids: {$keys: {as: set}}

# 谨慎使用:`$rows`(可能传递大量数据,并触发 rows barrier)
sources:
  customers:
    params:
      rows: {$rows: {cache_mode: batch}}

# 不推荐
main_source:
  params:
    api_key: "sk-xxxxx"

# 推荐:由调用方或 loader 在运行时注入(Scalim 不会自动做 `${...}` 插值)
main_source:
  params:
    api_key: null

import os

def load_orders(*, api_key=None, **kwargs):
    api_key = api_key or os.environ["API_KEY"]
    ...

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunSecurityOptions, run

# 模块级 allowlist(允许该模块及其子模块)
run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
    ),
)

# 函数级 allowlist(更精确,推荐)
run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(
            allowed_modules=frozenset(),
            allowed_functions=frozenset(
                [
                    "myapp.loaders:load_orders",
                    "myapp.loaders:load_customers",
                ]
            ),
        ),
    ),
)

main_source:
  source_id: orders
  loader: ".loaders:load_orders"  # => myapp.reports.loaders:load_orders

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunSecurityOptions, compile

# 推荐:显式 allowlist(安全默认)
_ = compile(
    "path/to/config.yaml",
    options=DemandRunOptions(security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"]))),
)

fields:
  # ❌ 不被允许: method call
  city:
    compute: "address.get('city', '')"

  # ✅ 推荐: call_by
  city:
    call_by: "myapp.transforms:dict_get(mapping=address, key='city', default='')"

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.relations import RelationConfig, RelationObserver

relations = RelationObserver(
    RelationConfig(
        sampling_rate=1.0,  # 采样率 100%
        log_type_mismatch=True,
        report_format="json",
        output_path="./debug_relations.json",
        include_details=True,
    )
)

_ = run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[relations]),
    ),
)

fields:
  order_amount:
    name: 订单金额
    compute: "quantity * unit_price"

  tax_amount:
    name: 税费
    compute: "order_amount * tax_rate"

sources:
  customers:
    fields:
      customer_name:
        extract: name
        name: 客户名称

  products:
    fields:
      product_name:
        extract: name
        name: 产品名称

# lookup_cast: 关联时转换
relations:
  orders_to_customers:
    steps:
      - from: orders.customer_id
        to: customers.customer_id
        lookup_cast:
          int: {}      # 确保 orders.customer_id 是整数

# value_cast: 输出时转换
main_source:
  fields:
    amount:
      value_cast: decimal    # 金融/金额类字段推荐: 用 Decimal 避免 float 精度问题

sources:
  # 适用:订单类型(数据量小)
  order_types:
    cache_mode: preload_forever

  # 不适用:订单明细(数据量大)
  order_details:
    cache_mode: none

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run

run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp"])),
        runtime=DemandRunRuntimeOptions(batch_size=500),  # 默认 1000
    ),
)

resources:
  files:
    detail_csv:
      csv_file:
        path: ./output

outputs:
  - name: detail
    to: {file: detail_csv}
    fields: [order_id]

sources:
  customers:
    params:
      ids: {$keys: {as: set}}

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.ob.presets.memory import MemoryOptimizationObserver

mem = MemoryOptimizationObserver(auto_report=True)
_ = run(
    "path/to/report.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[mem]),
    ),
)

# 使用 CLI 验证
scalim-cli yaml-dsl validate config.yaml

scalim-cli yaml-dsl validate --json config.yaml

# 或在代码中(编译时会自动校验;需配置 allowlist)
from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunSecurityOptions, compile

_ = compile(
    "config.yaml",
    options=DemandRunOptions(security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"]))),
)

resources:
  books:
    report:
      xlsx:
        path: ./output

outputs:
  - name: detail
    to: {book: report, sheet: 明细}
    fields: [order_id]

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunRuntimeOptions, DemandRunSecurityOptions, run
from scalim.hooks.base import BaseHook


class _MyHook(BaseHook):
    def on_pipeline_end(self, event) -> None:  # type: ignore[override]
        _ = event


# 执行配置
result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        runtime=DemandRunRuntimeOptions(components=[_MyHook()]),  # (可选) 附加自定义 hook/observer 组件
    ),
)

# 访问结果
print(f"处理行数: {result.total_rows}")
print(f"输出路径: {result.output_path}")

from scalim.dsl.yaml_dsl import DemandRunOptions, DemandRunSecurityOptions, DemandRunTemplateOptions, run

result = run(
    "path/to/config.yaml",
    options=DemandRunOptions(
        security=DemandRunSecurityOptions(allowed_modules=frozenset(["myapp.loaders"])),
        template=DemandRunTemplateOptions(
            template_vars={"out_root": "./out"},
            rendered_yaml_max_len=2_000_000,  # (可选)按需调大
        ),
    ),
)