Dumb Q1: In this toy example,

paginated_data

returns

List[Dict[str, Any]]

such as could be dropped into a

pd.DataFrame

. The result

data

is

{"paginated_data": [{… }

. Can I return a df without having to do something like

pd.DataFrame(data['paginated_data'])

? It matters because

final_vars

may be determined at runtime based on something like tags. Sure I've done this before but I forget…

dr = (
    await async_driver.Builder()
    .with_config(config)
    .with_modules(transforms)
    .build()
)
data = await dr.execute(
    ["paginated_data"], inputs={"endpoint": endpoint, "request_params": request_params}
)

… and Q2: Suppose

paginated_data

is something like the following and we want to reuse the transform across various dataflows assembled dynamically using module composition, would it be better to try and write this as a materializer?

async def paginated_data(
    endpoint: str,
    request_params: Dict[str, Any],
    *,
    api_settings: settings.Settings,
    pagination_request: schemas.PaginationRequest,
    pagination_params: Optional[schemas.PaginationParams] = None,
) -> List[Dict[str, Any]]:
    pagination_params = pagination_params or schemas.PaginationParams()
    async with pagination.PaginationService(api_settings) as pagination_service:
        result = await pagination_service.fetch_all_pages(
            endpoint,
            pagination_request=pagination_request,
            pagination_params=pagination_params,
            extra_params=request_params,
        )

    return result

… and Q3: Suppose that in some cases we want to do a groupby operation on the dataframe and apply a lightly configurable set of transforms to each group, what's the best pattern for that to preserve visibility of the transforms applied to each group?

Hey! Some responses: 1. Yes, this is quite possible — if you use the right result builder (

result_builder = base.PandasDataFrameResult()

and pass it into

with_adapters(result_builder)

it’ll give you what you want! 2. You have a few ways of doing this a. A materializer makes a lot of sense here — it’s a common way of “loading” the data, so wrapping it in a class and tacking it on is good b. You can also make it in a module — E.G. a module called

data_loaders

that defines this, and then make subsequence modules declare

paginated_data

as parameters to their functions. IMO this is a really clean way to reuse that promotes code reuse + clearly organized naming. 3. I think you’re going to really appreciate the

@pipe_output

decorator (thanks @Jernj Frank for implementing!!). Given that it’s “lightly configurable”, this has a capability of having a per-step

when

condition, allowing you to pass in config data to the driver and use that to control the shape of the chain. Docs/code examples here: https://hamilton.dagworks.io/en/latest/reference/decorators/pipe/#pipe-output

Thanks @Elijah Ben Izzy, will give this a go today.

Yep! Let us know how much/if the issue above is a blocker for you, I think the workaround should help

Will do. QQ: When I do,

result_builder = PandasDataFrameResult()
dr = (
    await async_driver.Builder()
    .with_config(config)
    .with_modules(transforms)
    .with_adapters(result_builder)
    .build()
)
data = await dr.execute(
    ["paginated_data"], inputs={"endpoint": endpoint, "request_params": request_params}
)

I get

paginated_data
0	{'record_id': '0x57AE4108A24DD9E51468C03FA003A...
1	{'record_id': '0x3CDCFAD8388BD79054F69C71230CB...
2	{'record_id': '0x478481F7DBBE326F122B299DAE765...

which is what was confusing me before. It's the value of the

paginated_data

KV pair that needs to be dropped into the DF constructor. Maybe I can setup the transform differently? I think the issue is that the result builder envisages combining series but in this context I have data as records… if that makes sense?

Yeah, so I think this is probably the case for (a) turning it into a series to conform with what it expects or (b) creating a custom result builder

Okay. In the event I need to do the second option, is there an easy way of extracting the name of the final vars dynamically? In the past, when using module composition, I've done things like

@tag(data_product="final")

and some filtering to find the list to pass to

.execute(…

. But in this context I will want to select from various tag options at runtime and might need access the correct keys in a custom result builder.

So the result builder has

build_result

, which takes in

**outputs

— that’s a dict of the node name -> node returned value from the DAG

Okay thanks. Lm think about whether I unequivocally need to solve that problem before trying to do it.

👍