Dagster asset factory example

config.json

[
    {
        "dag_id": "powerbi1",
        "python_script": "some_script.py",
        "dataset_id": "fdsjl4539fdjsk"
    },
    {
        "dag_id": "powerbi2",
        "python_script": "some_other_script.py",
        "dataset_id": "89fdskfds0"
    },
    {
        "dag_id": "powerbi3",
        "python_script": "some_other_script.py",
        "dataset_id": "89fdfdsjklskfds0"
    }
]

generator.py

import json
from typing import Any, Mapping

from dagster import Definitions, asset


def refresh_dataset(dataset_id):
    """Do HTTP POST to PowerBI"""


def generate_powerbi_asset(powerbi_blob: Mapping[str, Any]):
    @asset(name=powerbi_blob["dag_id"])
    def _asset(context):
        context.log.info(powerbi_blob)
        refresh_dataset(powerbi_blob["dataset_id"])

    return _asset


json_specs = json.load(open("powerbi.json"))
assets = [generate_powerbi_asset(json_spec) for json_spec in json_specs]


defs = Definitions(assets=assets)