[14/n][dagster-airbyte] Implement materialization method for AirbyteCloudWorkspace

[maximearmstrong]

Summary & Motivation

This PR implements AirbyteCloudWorkspace.sync_and_poll, the materialization method for Airbyte Cloud assets. This method:

• calls AirbyteCloudClient.sync_and_poll
• takes the AirbyteOutput returned by AirbyteCloudClient.sync_and_poll and generates the asset materializations
• yields MaterializeResult for each expected asset and AssetMaterialization for each unexpected asset
  • a connection table that was not in the connection at definitions loading time can be in the AirbyteOutput. Eg. the table was added after definitions loading time and before sync.
• logs a warning for each unmaterialized table
  • a connection table can be created at definitions loading time, but can be missing in the AirbyteOutput. Eg. the table was deleted after definitions loading time and before sync.

Can be leveraged like:

from dagster_airbyte import AirbyteCloudWorkspace, airbyte_assets

import dagster as dg

airbyte_workspace = AirbyteCloudWorkspace(
    workspace_id=dg.EnvVar("AIRBYTE_CLOUD_WORKSPACE_ID"),
    client_id=dg.EnvVar("AIRBYTE_CLOUD_CLIENT_ID"),
    client_secret=dg.EnvVar("AIRBYTE_CLOUD_CLIENT_SECRET"),
)


@airbyte_assets(
    connection_id="airbyte_connection_id",
    name="airbyte_connection_id",
    group_name="airbyte_connection_id",
    workspace=airbyte_workspace,
)
def airbyte_connection_assets(context: dg.AssetExecutionContext, airbyte: AirbyteCloudWorkspace):
    yield from airbyte.sync_and_poll(context=context)


defs = dg.Definitions(
    assets=[airbyte_connection_assets],
    resources={"airbyte": airbyte_workspace},
)

How I Tested These Changes

Additional tests with BK

Changelog

[dagster-airbyte] Airbyte Cloud assets can now be materialized using the AirbyteCloudWorkspace.sync_and_poll(…) method in the definition of a @airbyte_assets decorator.

[maximearmstrong]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• [docs] Add docs for reworked Airbyte Cloud API #26595
• [docs] Move legacy Airbyte Cloud docs #26584
• [dagster-airbyte][docs] Update airbyte API docs with asset decorator, factory and materialization method #26583
• [14/n][dagster-airbyte] Implement materialization method for AirbyteCloudWorkspace #26559 👈 (View in Graphite)
• [13/n][dagster-airbyte] Implement airbyte_assets and build_airbyte_assets_definitions #26432
• [12/n][dagster-airbyte] Implement sync_and_poll method in AirbyteCloudClient #26431
• [11/n][dagster-airbyte] Implement cancel_job method in AirbyteCloudClient #26430
• [10/n][dagster-airbyte] Implement get_job_details method in AirbyteCloudClient #26429
• [9/n][dagster-airbyte] Implement base sync method in AirbyteCloudClient #26428
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

[maximearmstrong]

We add the translator to the metadata to reuse it in the materialization process.

[dpeng817]

I know the pattern predates this PR, but I feel like it's a pretty big antipattern. Perhaps I'm missing something, but I also don't see a subprocess actually being called anywhere? I don't see why we need to attach the translator to access it, why can't the translator just be an argument that you need to pass?

I guess the argument is that we do the same thing for dbt, likely fivetran, etc. but I think it's a pretty rough pattern that stretches our metadata concept beyond what's intended.

[dpeng817]

Also, kinda weird that we don't do this in the spec loader? Feels like there's a potential footgun there if a user for whatever reason decides to use the spec loader then provide to a multi asset, won't contain the translator

[maximearmstrong]

Also, kinda weird that we don't do this in the spec loader? Feels like there's a potential footgun there if a user for whatever reason decides to use the spec loader then provide to a multi asset, won't contain the translator

That's a good call. I agree that it should be set at the spec loader level. Currently, it is set at the asset spec level for dlt, but not for dbt and Sling - I will open a ticket about that.

I know the pattern predates this PR, but I feel like it's a pretty big antipattern. Perhaps I'm missing something, but I also don't see a subprocess actually being called anywhere? I don't see why we need to attach the translator to access it, why can't the translator just be an argument that you need to pass?

I think the main argument here - and for all other integrations - is how XWorkspace.sync_and_poll(...) is expected to be called, which is only in the context of the asset decorator.

Passing the translator to both the decorator and sync_and_poll function is a bit awkward, and would most likely be error-prone.

@airbyte_assets(
    connection_id="airbyte_connection_id",
    name="airbyte_connection_id",
    group_name="airbyte_connection_id",
    workspace=airbyte_workspace,
    dagster_airbyte_translator=CustomDagsterAirbyteTranslator(),
)
def airbyte_connection_assets(context: dg.AssetExecutionContext, airbyte: AirbyteCloudWorkspace):
    yield from airbyte.sync_and_poll(
        context=context, 
        dagster_airbyte_translator=CustomDagsterAirbyteTranslator() # potential errors here if another translator is passed
    )

That said, I agree that leveraging the metadata concept for that is kinda weird. I will open a ticket for that as well.

[maximearmstrong]

Moved to spec loader in 40057f9

[cmpadden]

FWIW the translator in metadata felt like anti-pattern when implementing the dlt integration too, but at that point the goal was trying to be consistent with dbt, and as Maxime said, specifying it in both locations is not ideal. Less ergonomic for the integration maintainer, more ergonomic for the user.

Just want to share that I have the same sentiment.

[dpeng817]

I don't love the pattern but I agree that to be consistent we probably want to add the translator to the spec metadata.

I think we need to be consistent and also do it in the spec loader.

q