[dagster-airbyte] Scaffold DagsterAirbyteTranslator for rework

python_modules/libraries/dagster-airbyte/dagster_airbyte/resources.py

@@ -20,13 +20,12 @@
 from dagster._annotations import experimental
 from dagster._config.pythonic_config import infer_schema_from_config_class
 from dagster._core.definitions.resource_definition import dagster_maintained_resource
-from dagster._record import record
-from dagster._serdes.serdes import whitelist_for_serdes
 from dagster._utils.cached_method import cached_method
 from dagster._utils.merger import deep_merge_dicts
 from pydantic import Field, PrivateAttr
 from requests.exceptions import RequestException
 
+from dagster_airbyte.translator import AirbyteWorkspaceData
 from dagster_airbyte.types import AirbyteOutput
 
 DEFAULT_POLL_INTERVAL_SECONDS = 10
@@ -801,42 +800,6 @@ def airbyte_cloud_resource(context) -> AirbyteCloudResource:
 # -------------
 
 
-@whitelist_for_serdes
-@record
-class AirbyteConnection:
-    """Represents an Airbyte connection, based on data as returned from the API."""
-
-    @classmethod
-    def from_connection_details(
-        cls,
-        connection_details: Mapping[str, Any],
-    ) -> "AirbyteConnection":
-        raise NotImplementedError()
-
-
-@whitelist_for_serdes
-@record
-class AirbyteDestination:
-    """Represents an Airbyte destination, based on data as returned from the API."""
-
-    @classmethod
-    def from_destination_details(
-        cls,
-        destination_details: Mapping[str, Any],
-    ) -> "AirbyteDestination":
-        raise NotImplementedError()
-
-
-@record
-class AirbyteWorkspaceData:
-    """A record representing all content in an Airbyte workspace.
-    This applies to both Airbyte OSS and Cloud.
-    """
-
-    connections_by_id: Mapping[str, AirbyteConnection]
-    destinations_by_id: Mapping[str, AirbyteDestination]
-
-
 @experimental
 class AirbyteCloudClient:
     """This class exposes methods on top of the Airbyte APIs for Airbyte Cloud."""

python_modules/libraries/dagster-airbyte/dagster_airbyte/translator.py

@@ -0,0 +1,56 @@
+from typing import Any, Mapping, NamedTuple
+
+from dagster._annotations import experimental
+from dagster._core.definitions.asset_spec import AssetSpec
+from dagster._record import record
+from dagster._serdes.serdes import whitelist_for_serdes
+
+
+class AirbyteConnectionTableProps(NamedTuple): ...
+
+
+@whitelist_for_serdes
+@record
+class AirbyteConnection:
+    """Represents an Airbyte connection, based on data as returned from the API."""
+
+    @classmethod
+    def from_connection_details(
+        cls,
+        connection_details: Mapping[str, Any],
+    ) -> "AirbyteConnection":
+        raise NotImplementedError()
+
+
+@whitelist_for_serdes
+@record
+class AirbyteDestination:
+    """Represents an Airbyte destination, based on data as returned from the API."""
+
+    @classmethod
+    def from_destination_details(
+        cls,
+        destination_details: Mapping[str, Any],
+    ) -> "AirbyteDestination":
+        raise NotImplementedError()
+
+
+@record
+class AirbyteWorkspaceData:
+    """A record representing all content in an Airbyte workspace.
+    This applies to both Airbyte OSS and Cloud.
+    """
+
+    connections_by_id: Mapping[str, AirbyteConnection]
+    destinations_by_id: Mapping[str, AirbyteDestination]
+
+
+@experimental
+class DagsterAirbyteTranslator:
+    """Translator class which converts a `AirbyteConnectionTableProps` object into AssetSpecs.
+    Subclass this class to implement custom logic for Airbyte content.
+    """
+
+    def get_asset_spec(self, props: AirbyteConnectionTableProps) -> AssetSpec:
+        """Get the AssetSpec for a table synced by an Airbyte connection."""
+        raise NotImplementedError()