Do you have an example of applying

@subdag

to a

Parallelizable

node? I am trying to do the following but and missing some of the plumbing.

what is

RAW_COLS_TO_EXTRACT

?

This is the corrected code, But here, where can I apply the

subdag

?

@config.when(split_by_dimension=True)
def split_trades_by_dimension(trades: pd.DataFrame) -> Parallelizable[pd.DataFrame]:
    """Group trades by dimension."""
    for dim, cols in DIMENSION_SCHEMES.items():
        grp = (
            trades
            .pipe(_create_portfolio_name_dimensions, columns=cols)
            .groupby(["portfolio_name"] + cols, as_index=False, dropna=False)
        )
        for _, dimension_df in grp:
            yield dimension_df.assign(dimension=dim)

@config.when(split_by_dimension=True)
def dimension_df(split_trades_by_dimension: pd.DataFrame) -> pd.DataFrame:
    """Extract dimension data from formatted positions."""
    return split_trades_by_dimension


@config.when(split_by_dimension=True)
@extract_columns(*
    ["portfolio_name"]
    + RAW_COLS_TO_EXTRACT
)
def trades_by_dimension(dimension_df: Collect[pd.DataFrame]) -> pd.DataFrame:
    """Extract dimension data from formatted positions."""
    return pd.concat(dimension_df, ignore_index=True)

It could depend on dimension_df

extract_columns

is needed for one of the

load_from=*modules

that I would like to apply to each

dimension_df

@Seth Stokes the way I would break it down is by working on the "inner" graphs independently, and then wiring them into larger graphs. E.g. work on the graph that you want to "book end" with parallelizeable and collect first independently. Once you have that graph working for an individual case, it should be straightforward to then apply parallelizable and collect. I think you're trying to combine too many decorators at once. Otherwise I am trying to read into what you want, and came up with the following structure from the code above:

def split_trades_by_dimension(trades: pd.DataFrame) -> Parallelizable[pd.DataFrame]:
    """bookend start"""
    for dim, cols in DIMENSION_SCHEMES.items():
        grp = (
            trades
            .pipe(_create_portfolio_name_dimensions, columns=cols)
            .groupby(["portfolio_name"] + cols, as_index=False, dropna=False)
        )
        for dimension_df in grp:
            yield dimension_df

@extract_columns(*
    ["portfolio_name"]
    + RAW_COLS_TO_EXTRACT
)
def dimension_df(split_trades_by_dimension: pd.DataFrame) -> pd.DataFrame:  # if this is important you could move this function into the subdag module, or a separate one rather than have it here
    """Extract dimension data from formatted positions."""
    return split_trades_by_dimension


@subdag(prepare_trades_for_request_obj, 
   inputs={
     input_name: source(input_name)
     for input_name in  ["portfolio_name", "dimension_df"] + RAW_COLS_TO_EXTRACT  
   }
)
def trades_by_dimension(portfolios_validated: pd.DataFrame) -> pd.DataFrame:: # coming from prepare_trades_for_request_obj
    """Extract dimension data from formatted positions."""
    return portfolios_validated

def all_trades_by_dimendion(trades_by_dimension: Collect[pd.DataFrame]) -> pd.DataFrame:
    # bookend end
    return pd.concat(portfolios_validated, ignore_index=True)

Thanks for this. I think what you did should work. Also, good to hear that iterating and crating a new module to add to the pipeline is the preferred method