Hi there, when I try to run

dagster dev

I get the following error

/home/byteuser/venv/lib/python3.10/site-packages/dagster/_core/workspace/context.py:591: UserWarning: Error loading repository location hillsg_dagster:dagster._core.errors.DagsterInvalidDefinitionError: "src_airtable__hg_data_entry/sub_learning_areas" is not a valid name in Dagster. Names must be in regex ^[A-Za-z0-9_]+$.

Looking into it, the Stream name for an Airbyte connection with Airtable is

"src_airtable__hg_data_entry/sub_learning_areas"

I can't control the stream name, this is how Airbyte resolves the Base/Table name with the Airbyte connector. When the table is created in the DB, it changes "/" to "_". Any idea how can I work around this issue?

Ran into something similar dagster airbyte doesn't respect the changes airbyte does to naming of special characters in field names either.

You might want to pass a custom _`connection_to_asset_key_fn`_ when loading assets?

Hi @Adam Bloom, how would the custom function look? The table db names are: src_airtable__hg_dat__ry_sub_learning_areas src_airtable__hg_data_entry_absence_groups This is what I got for now..

airbyte_assets = load_assets_from_airbyte_instance(
    AIRBYTE_INSTANCE,
    key_prefix=['sources'],
    connection_to_asset_key_fn=?)

Hi @Ohad, were you able to find a solution to this issue?

Unfortunately not @Hiroto Yamakawa pleas let me know if you do find a solution for it 🙏

Did you explore adding a custom mapping function? I don't see why that wouldn't work for your case. I'm not using one myself, so I don't have an example, but I'll link you to the default one if you haven't looked at that already

Great idea. Thanks, I’ll give it a go.

@Ohad the only way I was able to do it was as follow:

from dagster_airbyte.asset_defs import _clean_name

def connection_to_asset_key_function(meta, name):

    for item in meta.stream_data:
        item['stream']['name'] = _clean_name(item['stream']['name'])
    return AssetKey([meta.name, name])

and then in

dagster_airbyte/asset_defs.py

( a file from the package):

def parse_stream_tables(
    self, return_normalization_tables: bool = False
) -> Mapping[str, AirbyteTableMetadata]:
   [...]

    for stream in enabled_streams:
        name = _clean_name(cast(str, stream.get("stream", {}).get("name")))
   [...]

that way, the files with rejected charaters were loaded into Dagster

Thank you @Hiroto Yamakawa for sharing this, and it would be great if this could be merged into dagster_airbyte 🙏