Dagster dependencies logic is very confused for me

[eromoe]

I just want to assign dependencies for many cmd jobs, but it is super hard and ambiguous.

I create some funcs to automaulticaly create bulk ops

from .cmd import (
    create_cmd_download_ops, 
    create_cmd_chain_ops, 
    create_cmd_parallel_ops
)

Use as op /job

@op
def etf_kline():
    create_cmd_download_ops([
        r'''
        conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\tushare_api\etf.py
        '''
    ])
    
@op
def etf_adj():
    create_cmd_download_ops([
        r'''
        conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\tushare_api\fund_adj.py
        '''
    ])
        
@op
def etf_qfq(etf_kline, etf_adj):
    create_cmd_parallel_ops({
        # stock 后复权 前复权
        "etf_qfq": "conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\etl.py fq2 -t etf -a qfq",
    })

@op
def etf_hfq(etf_kline, etf_adj):
    create_cmd_parallel_ops({
        # stock 后复权 前复权
        "etf_hfq": "conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\etl.py fq2 -t etf -a hfq",
    })

@job
def etf_job():
    a = etf_kline()
    b = etf_adj()
    etf_qfq(a, b)
    etf_hfq(a, b)

Dependancies is correct:

But can't no run ( said etf_kline need context, but directly run in a job is fine )

Use as graph /job

@graph
def etf_kline_adj_job():
    create_cmd_parallel_ops({
        "etf_kline": "conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\tushare_api\etf.py",
        "etf_factor": "conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\tushare_api\fund_adj.py",
    })

@graph
def etf_fq_job(etf_kline_adj_job):
    create_cmd_parallel_ops({
        # stock 后复权 前复权
        "etf_hfq": "conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\etl.py fq2 -t etf -a hfq",
        "etf_qfq": "conda activate quant && python E:\Workspace\github_me\data_dumper\data_dumper\etl.py fq2 -t etf -a qfq",
    })
    

@job
def etf_job():
    etf_fq_job(etf_kline_adj_job())

Can not add dependencies. Error:

dagster._core.errors.DagsterInvalidDefinitionError: @graph 'etf_fq_job' has unmapped input 'etf_kline_adj_job'. Remove it or pass it to the appropriate op/graph invocation.

Can not create dependencies between graphs .

Why I don't tend to write an op for every cmd ?

Because I am lazy and obsessive, tend to use a function group all relative jobs , so that no redundant code and orgnization pretty.
I really hope there is similar stuff like -> in airflow, to link dependency jobs without input and output.