Getting all partition keys from an upstream asset in a downstream unpartitioned asset

[nyte-owl]

I have an upstream asset partitioned with a DynamicPartitionsDefinition.

In my downstream asset, I'd like to get a list of all the partition keys from the upstream asset. The problem I'm running into is an error message from dagster saying "The input has no asset partitions".

Example code:

unique_id_partitions = dg.DynamicPartitionsDefinition(name="unique_id_partitions")



@dg.asset(
    name="Upstream Asset",
    partitions_def=unique_id_partitions,
)
def upstream_asset(context):
    context.log("This is the upstream asset.")

# < sensor and job here to create run requests for upstream_asset with unique ID for partition based on some external data >

@dg.asset(
    name="Downstream Asset",
    deps=[upstream_asset],
)
def downstream_asset(context):
    # This line causes the error:
    unique_ids = context.asset_partition_keys_for_input("upstream_asset")

Running dagster locally with version 1.10.19.

I'm able to workaround this by using a different approach:

instance = context.instance
unique_ids = unique_id_partitions.get_partition_keys(dynamic_partitions_store=instance)

But I'm wondering why the first approach doesn't work right? Does the upstream asset need to be an arg input to the downstream asset instead of non arg deps?