A related question...Can I have another upstream asset that creates the DynamicPartitions or do I need to use a sensor like in the docs?

you could add dynamic partitions in another asset e.g.:

@asset
def adds_partitions(context):
    context.instance.add_dynamic_partitions(partitions_def_name, partition_keys)
    ....

One caveat here is that unexpected behaviors may occur if you mutate the dynamic partitions def during run for one of its partitions: i.e. if you added partitions in an upstream dynamic-partitioned asset, and attempted to read all partitions in in a downstream unpartitioned asset in the same run. So I would recommend not mutating the partitions def in one of its runs.

is that right?

Can Dynamic Partitions be Multipartitions?

Looks like a no, but I just want to confirm, and I'm wondering if it will be supported in the future. Thanks!

Yes I see it, that's awesome. In that case, @claire, how would I have the downstream asset take all partitions of the one of the dimensions? What I mean is that the Multipartition would have 1 dimension that is Dynamic, say

name

. And the other one is

year

. Then the downstream asset would be using the partition

year

and combining all the

names

. Does that work automatically, and would the input asset also look like a dict with the

names

as the keys? Thanks!

In terms of what the IO manager returns, it depends on the IO manager. For the default

fs_io_manager

, you would get a dict containing the partition key path to the object. Currently the path orders each dimension in a different folder, if you wanted to get back the multipartition key to the object that would look like this:

@asset(partitions_def=DailyPartitionsDefinition("2023-01-01"))
def downstream_of_multipartitions_asset(upstream):
    def _get_key_from_multipartition_path(path: str) -> MultiPartitionKey:
        dimension_names = sorted(multipartitions_def.partition_dimension_names)
        keys_ordered_by_dimension = path.split("/")
        return MultiPartitionKey(
            {tup[0]: tup[1] for tup in list(zip(dimension_names, keys_ordered_by_dimension))}
        )

    object_by_key = {_get_key_from_multipartition_path(path): object for path, object in upstream.items()}
    ...

Not super ergonomic at the moment, I'd recommend filing an issue to instead return a mapping of multipartition key -> object instead of the filepath dict

@Rubén Briones https://github.com/dagster-io/dagster/blob/master/CHANGES.md#new-1 • (experimental) Dynamic partitions definitions can now exist as dimensions of multi-partitions definitions.

Hi @claire, sorry I have another question. I'm looking at adding dynamic partitions in an asset instead of sensor. I'm wondering if this still applies for the asset context

context.instance.add_dynamic_partitions

yes, verified

These should be multipartitions where the other partition with the metric is Year

Ideally I'd like to have partitions only materialize if they're missing in that job.