How can I implement dynamic partitions in sequence in the processing of files and chunks.

[vlreinier]

I have the following code example:

partition_a = DynamicPartitionsDefinition(name="partition_a")
partition_b = DynamicPartitionsDefinition(name="partition_b")


@dg.asset
def files(context: AssetExecutionContext) -> dict:
    """Creates partitions dynamically for files."""
    data = {"file_1": "content_1", "file_2": "content_2"}

    # Dynamically register files as partitions
    context.instance.add_dynamic_partitions(
        partitions_def_name="partition_a",
        partition_keys=list(data.keys()),
    )

    return data  # {file_1: content_1, file_2: content_2}


@dg.asset(partitions_def=partition_a)
def file_chunks(context: AssetExecutionContext, files) -> dict:
    """Processes a file partition into multiple chunks."""
    file_key = context.partition_key  # Example: "file_1"
    file_content = files[file_key]  # Get file data

    # Simulate breaking file into chunks
    chunks = {f"{file_key}_chunk_1": "chunk_data_1", f"{file_key}_chunk_2": "chunk_data_2"}

    # Dynamically register chunk partitions
    context.instance.add_dynamic_partitions(
        partitions_def_name="partition_b",
        partition_keys=list(chunks.keys()),
    )

    return chunks  # Example: {"file_1_chunk_1": "chunk_data_1", "file_1_chunk_2": "chunk_data_2"}


@dg.asset(partitions_def=partition_b)
def processed_file_chunk(context: AssetExecutionContext, file_chunks) -> None:
    """Processes chunks dynamically."""
    chunk_key = context.partition_key  # Example: "file_1_chunk_1"

This result in the following error dagster._core.errors.DagsterInvariantViolationError: Asset partition AssetKeyPartitionKey(asset_key=AssetKey(['processed_file_chunk']), partition_key='file_1_chunk_2') depends on invalid partition keys {AssetKeyPartitionKey(asset_ke y=AssetKey(['file_chunks']), partition_key='file_1_chunk_2')}.

What is the correct way of implementing this with partitioning?