dagster-pipe-typescript

排字程序语言的管道执行 .

允许任何打字程序与 Dagster 管弦乐器之间的融合 .

使用量

先决条件

• 安装节点和 npm
• 安装打字机编译器( e)npm install -g typescript(中文(简体) )

安装

@dagster-io/dagster-pipes 作为 npm 软件包可用 :

npm install @dagster-io/dagster-pipes

示例

除了下面的例子之外,您还可以找到一个带有打字管道资产的示例 example-project/ 目录.

安装后 @dagster-io/dagster-pipes npm 软件包, 您的打字程序可以导入它与 dagster 集成 .

import * as dagster_pipes from '@dagster-io/dagster-pipes';

using context = dagster_pipes.openDagsterPipes()

context.logger.info("this is a log message from the external process")
context.logger.warning("This is an example warning from the external process")

context.reportAssetMaterialization(
    {
        "row_count": 100
    }
)
context.reportAssetCheck(
    "example_typescript_check", true
)

在达克斯特管弦乐法中,增加以下内容:

import subprocess
from pathlib import Path
import dagster as dg

@dg.asset(
    check_specs=[
        dg.AssetCheckSpec(
            name="example_typescript_check",
            asset="example_typescript_asset"
        )
    ]
)
def example_typescript_asset(
    context: dg.AssetExecutionContext, 
    pipes_subprocess_client: dg.PipesSubprocessClient
) -> dg.MaterializeResult:
    external_script_path = dg.file_relative_path(__file__, "../external_typescript_code/")
    
    subprocess.run(["npm install"],
                    cwd=external_script_path).check_returncode()
    subprocess.run(["tsc"],
                   cwd=external_script_path).check_returncode()

    return pipes_subprocess_client.run(
        command=["node", Path(external_script_path)],
        context=context,
    ).get_materialize_result()


defs = dg.Definitions(
    assets=[example_typescript_asset],
    resources={
        "pipes_subprocess_client": dg.PipesSubprocessClient()
    },
)

发展

正在编译开发版本

npm install && npm run build

测试

npm test
uv run --all-extras pytest

线性

npm run lint