[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

• [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.