using @multi_observable_source_asset with daily partitions

[harikaduyu]

I'd like to learn how I can use @multi_observable_source_asset decorator with daily partitions. I have a bunch of tables dumped to S3 by an external resource in the format path/to/{table_name}_{YYY-MM-DD}.csv. I have defined a list of assets using @multi_asset decorator which reads those S3 files and dumps it to somewhere else.

TABLES = {
    "table_1": {
        "columns": [
            {
                "name": "column 1",
                "type": "INTEGER",
                "description": "some description",
            },
        ]
    },
    "table_2": {
        "columns": [
            {
                "name": "column 2",
                "type": "STRING",
                "description": "some description",
            },
        ]
    },
}

daily_partition_definition = DailyPartitionsDefinition(start_date="2024-03-12")


@multi_observable_source_asset(
    specs=[
        AssetSpec(
            key=["some", "external_asset", table],
            group_name="group",
        )
        for table in TABLES.keys()
    ],
    can_subset=True,
    partitions_def=daily_partition_definition,
)
def observable_assets(context: AssetExecutionContext, s3: S3Resource):
    client = s3.get_client()
    for table in context.selected_output_names:
        file = context.asset_key_for_output(table).path[-1]
        file_path = f"path/to/{file}_{context.partition_key}.csv"
        try:
            obj = client.get_object(
                Bucket=BUCKET,
                Key=file_path,
            )
            yield ObserveResult(
                asset_key=AssetKey(["some", "external_asset", file]),
                data_version=DataVersion(obj["LastModified"].strftime(format="%Y-%m-%dT%H:%M:%S")),
            )
        except BaseException as e:
            context.log.warn(f"Failed to fetch file {file_path}")
            context.log.warn(e)
            return None
            
@multi_asset(
    outs={
        table: AssetOut(
            key=AssetKey([*PREFIX, f"src_{table}"]),
            is_required=False,
            io_manager_key="gcs_io_manager",
            auto_materialize_policy=AutoMaterializePolicy.eager(),
        )
        for table in TABLES.keys()
    },
    deps=[AssetKey(["some", "external_asset", table]) for table in TABLES.keys()],
    internal_asset_deps={
        table: {AssetKey(["some", "external_asset", table])} for table in TABLES.keys()
    },
    partitions_def=daily_partition_definition,
    can_subset=True,
)
def src_asset_func(
    context: AssetExecutionContext,
    s3: S3Resource,
):
    client = s3.get_client()
    for output_name in context.selected_output_names:
        file = context.asset_key_for_output(output_name).path[-1].replace("src_", "")
        file_path = f"path/to/{file}_{context.partition_key}.csv"
        obj = client.get_object(
            Bucket=BUCKET,
            Key=file_path,
        )
        in_buffer = io.BytesIO(obj["Body"].read())
        df = pd.read_csv(in_buffer)
        yield Output(
            value=df,
            output_name=output_name,
        )

The assets that are defined by @multi_asset decorator works fine but I'm confused how to observe the external assets with a schedule and partitions (and partitions defined in the file name) and materialise the downstream asset partitions if there's an observation for that partition accordingly. When I try to run an observation I get the following error:

dagster._core.errors.DagsterInvariantViolationError: Cannot access partition_key for a non-partitioned run

Can anybody help me with this? I'm open for ideas, alternatives and feedback. I guess, I can also accomplish this using external assets and a sensor, instead of using an observation schedule.