如何通过API传递参数给Dagster资产并触发作业运行？

1. 先调整资产和作业的定义 #

你现在写的资产直接传player_name参数的方式不符合Dagster的资产逻辑——Dagster资产默认不直接接收外部参数，得通过配置或者作业来传递。这里给你两种可行的调整方式：

方式一：用配置传递参数 #

先定义一个配置类，让资产从配置里读取player_name：

from dagster import asset, Config, get_dagster_logger

class PlayerConfig(Config):
    player_name: str

@asset
def player_already_registered(config: PlayerConfig):
    logger = get_dagster_logger()
    logger.info(f"检查玩家 {config.player_name} 的注册状态")
    
    q = text(
        '''
            SELECT 
                COUNT(*)
            FROM
                `player_account_info`
            WHERE
                summonerName = :player_name
        '''
    )

    result = database.conn.execute(q, player_name=config.player_name).fetchone()[0]
    return bool(result)

然后把这个资产打包成一个作业，方便后续触发：

from dagster import define_asset_job

check_player_job = define_asset_job(
    name="check_player_registration_job",
    selection=["player_already_registered"]
)

2. 搭建API端点触发作业 #

假设你用FastAPI来写这个端点（用Flask逻辑也类似），接收GET参数里的player_name，然后调用Dagster的Python API触发作业：

先装依赖 #

pip install fastapi uvicorn dagster

写API代码 #

from fastapi import FastAPI
from dagster import DagsterInstance, execute_job
# 这里替换成你存放Dagster资产的模块名
from your_dagster_code import check_player_job

app = FastAPI()

@app.get("/check-player-reg")
def check_player(player_name: str):
    # 初始化Dagster实例
    instance = DagsterInstance.get()
    
    # 传配置触发作业运行
    run_result = execute_job(
        check_player_job,
        instance=instance,
        run_config={
            "ops": {
                "player_already_registered": {
                    "config": {
                        "player_name": player_name
                    }
                }
            }
        }
    )
    
    # 返回结果
    return {
        "玩家名称": player_name,
        "是否已注册": run_result.output_for_node("player_already_registered"),
        "运行状态": "成功" if run_result.success else "失败"
    }

3. 启动服务 #

1. 先启动Dagster开发服务：

dagster dev

2. 再启动FastAPI服务：

uvicorn your_api_file:app --reload

现在访问 http://localhost:8000/check-player-reg?player_name=你的玩家名 就能触发资产运行，同时拿到结果。

备选方案：用Dagster自带的GraphQL API触发 #

如果你不想自己写API，也可以直接用Dagster自带的GraphQL接口来触发作业。比如用curl发送请求：

curl -X POST http://localhost:3000/graphql \
-H "Content-Type: application/json" \
-d '{
  "query": "mutation($runConfig: RunConfigInput!, $jobName: String!) { launchRun(runConfig: $runConfig, jobName: $jobName) { runId } }",
  "variables": {
    "jobName": "check_player_registration_job",
    "runConfig": {
      "ops": {
        "player_already_registered": {
          "config": {
            "player_name": "测试玩家"
          }
        }
      }
    }
  }
}'

触发后可以在Dagster UI里查看运行结果，或者再用GraphQL查询返回的runId获取详细输出。

内容的提问来源于stack exchange，提问作者NFeruch - FreePalestine