Supper filters in asset factory and decorator

Author: maximearmstrong

python_modules/libraries/dagster-fivetran/dagster_fivetran/asset_decorator.py

@@ -3,7 +3,7 @@
 from dagster import AssetsDefinition, multi_asset
 from dagster._annotations import beta
 
-from dagster_fivetran.resources import FivetranWorkspace
+from dagster_fivetran.resources import FivetranFilter, FivetranWorkspace
 from dagster_fivetran.translator import DagsterFivetranTranslator, FivetranMetadataSet
 
 
@@ -15,6 +15,7 @@ def fivetran_assets(
     name: Optional[str] = None,
     group_name: Optional[str] = None,
     dagster_fivetran_translator: Optional[DagsterFivetranTranslator] = None,
+    fivetran_filter: Optional[FivetranFilter] = None,
 ) -> Callable[[Callable[..., Any]], AssetsDefinition]:
     """Create a definition for how to sync the tables of a given Fivetran connector.
 
@@ -27,6 +28,7 @@ def fivetran_assets(
         dagster_fivetran_translator (Optional[DagsterFivetranTranslator], optional): The translator to use
             to convert Fivetran content into :py:class:`dagster.AssetSpec`.
             Defaults to :py:class:`DagsterFivetranTranslator`.
+        fivetran_filter (Optional[FivetranFilter]): Filters the set of Fivetran objects to fetch.
 
     Examples:
         Sync the tables of a Fivetran connector:
@@ -102,6 +104,7 @@ def fivetran_connector_assets(context: dg.AssetExecutionContext, fivetran: Fivet
 
     """
     dagster_fivetran_translator = dagster_fivetran_translator or DagsterFivetranTranslator()
+    fivetran_filter = fivetran_filter or FivetranFilter(connector_ids=frozenset({connector_id}))
 
     return multi_asset(
         name=name,
@@ -110,7 +113,8 @@ def fivetran_connector_assets(context: dg.AssetExecutionContext, fivetran: Fivet
         specs=[
             spec
             for spec in workspace.load_asset_specs(
-                dagster_fivetran_translator=dagster_fivetran_translator
+                dagster_fivetran_translator=dagster_fivetran_translator,
+                fivetran_filter=fivetran_filter,
             )
             if FivetranMetadataSet.extract(spec.metadata).connector_id == connector_id
         ],

python_modules/libraries/dagster-fivetran/dagster_fivetran/asset_defs.py

@@ -32,7 +32,12 @@
 from dagster._utils.log import get_dagster_logger
 
 from dagster_fivetran.asset_decorator import fivetran_assets
-from dagster_fivetran.resources import DEFAULT_POLL_INTERVAL, FivetranResource, FivetranWorkspace
+from dagster_fivetran.resources import (
+    DEFAULT_POLL_INTERVAL,
+    FivetranFilter,
+    FivetranResource,
+    FivetranWorkspace,
+)
 from dagster_fivetran.translator import (
     DagsterFivetranTranslator,
     FivetranConnectorTableProps,
@@ -733,6 +738,7 @@ def build_fivetran_assets_definitions(
     *,
     workspace: FivetranWorkspace,
     dagster_fivetran_translator: Optional[DagsterFivetranTranslator] = None,
+    fivetran_filter: Optional[FivetranFilter] = None,
 ) -> Sequence[AssetsDefinition]:
     """The list of AssetsDefinition for all connectors in the Fivetran workspace.
 
@@ -741,6 +747,7 @@ def build_fivetran_assets_definitions(
         dagster_fivetran_translator (Optional[DagsterFivetranTranslator], optional): The translator to use
             to convert Fivetran content into :py:class:`dagster.AssetSpec`.
             Defaults to :py:class:`DagsterFivetranTranslator`.
+        fivetran_filter (Optional[FivetranFilter]): Filters the set of Fivetran objects to fetch.
 
     Returns:
         List[AssetsDefinition]: The list of AssetsDefinition for all connectors in the Fivetran workspace.
@@ -807,9 +814,10 @@ def get_asset_spec(self, props: FivetranConnectorTableProps) -> dg.AssetSpec:
 
     """
     dagster_fivetran_translator = dagster_fivetran_translator or DagsterFivetranTranslator()
+    fivetran_filter = fivetran_filter or FivetranFilter()
 
     all_asset_specs = workspace.load_asset_specs(
-        dagster_fivetran_translator=dagster_fivetran_translator
+        dagster_fivetran_translator=dagster_fivetran_translator, fivetran_filter=fivetran_filter
     )
 
     connector_ids = {
@@ -826,6 +834,7 @@ def get_asset_spec(self, props: FivetranConnectorTableProps) -> dg.AssetSpec:
             name=connector_id,
             group_name=connector_id,
             dagster_fivetran_translator=dagster_fivetran_translator,
+            fivetran_filter=fivetran_filter,
         )
         def _asset_fn(context: AssetExecutionContext, fivetran: FivetranWorkspace):
             yield from fivetran.sync_and_poll(context=context)

python_modules/libraries/dagster-fivetran/dagster_fivetran/resources.py

@@ -921,7 +921,9 @@ def fetch_fivetran_workspace_data(
         fivetran_filter: Optional[FivetranFilter] = None,
     ) -> FivetranWorkspaceData:
         """Retrieves all Fivetran content from the workspace and returns it as a FivetranWorkspaceData object.
-        fivetran_filter (Optional[FivetranFilter]): Filters the set of Fivetran objects to fetch.
+
+        Args:
+            fivetran_filter (Optional[FivetranFilter]): Filters the set of Fivetran objects to fetch.
 
         Returns:
             FivetranWorkspaceData: A snapshot of the Fivetran workspace's content.