[dagster-fivetran]: Add op_tags to build_fivetran_assets (#12474)

### Summary & Motivation
Dagster supports the addition of custom k8s configurations to ops and
jobs via tags. In an environment where a custom configuration is
required to be attached e.g. Kubernetes Pods, the current
`build_fivetran_assets` can't support attaching these tags to the
underlying Op, causing the job to fail.

In this PR I'm adding an optional `op_tags` parameter to support use
case

### How I Tested These Changes
Adding a unit test to asset op_tags are as expected

---------

Co-authored-by: Ben Pankow <[email protected]>

Author: Sedosa

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

@@ -43,6 +43,7 @@ def _build_fivetran_assets(
     resource_defs: Optional[Mapping[str, ResourceDefinition]] = None,
     group_name: Optional[str] = None,
     infer_missing_tables: bool = False,
+    op_tags: Optional[Mapping[str, Any]] = None,
 ) -> Sequence[AssetsDefinition]:
     asset_key_prefix = check.opt_sequence_param(asset_key_prefix, "asset_key_prefix", of_type=str)
 
@@ -69,6 +70,7 @@ def _build_fivetran_assets(
         compute_kind="fivetran",
         resource_defs=resource_defs,
         group_name=group_name,
+        op_tags=op_tags,
     )
     def _assets(context):
         fivetran_output = context.resources.fivetran.sync_and_poll(
@@ -118,6 +120,7 @@ def build_fivetran_assets(
     metadata_by_table_name: Optional[Mapping[str, MetadataUserInput]] = None,
     group_name: Optional[str] = None,
     infer_missing_tables: bool = False,
+    op_tags: Optional[Mapping[str, Any]] = None,
 ) -> Sequence[AssetsDefinition]:
     """
     Build a set of assets for a given Fivetran connector.
@@ -147,6 +150,10 @@ def build_fivetran_assets(
             in destination_tables even if they are not present in the Fivetran sync output. This is useful
             in cases where Fivetran does not sync any data for a table and therefore does not include it
             in the sync output API response.
+        op_tags (Optional[Dict[str, Any]]):
+             A dictionary of tags for the op that computes the asset. Frameworks may expect and
+             require certain metadata to be attached to a op. Values that are not strings will be
+             json encoded and must meet the criteria that json.loads(json.dumps(value)) == value.
 
     **Examples:**
 
@@ -193,6 +200,7 @@ def build_fivetran_assets(
         metadata_by_table_name=metadata_by_table_name,
         group_name=group_name,
         infer_missing_tables=infer_missing_tables,
+        op_tags=op_tags,
     )
 
 

python_modules/libraries/dagster-fivetran/dagster_fivetran_tests/test_asset_defs.py

@@ -56,7 +56,8 @@ def test_fivetran_group_label(group_name, expected_group_name):
         (["schema1.tracked", "does.not_exist"], True, False),
     ],
 )
-def test_fivetran_asset_run(tables, infer_missing_tables, should_error, schema_prefix):
+@pytest.mark.parametrize("op_tags", [None, {"key1": "value1"}])
+def test_fivetran_asset_run(tables, infer_missing_tables, should_error, schema_prefix, op_tags):
     ft_resource = fivetran_resource.configured({"api_key": "foo", "api_secret": "bar"})
     final_data = {"succeeded_at": "2021-01-01T02:00:00.0Z"}
     api_prefix = f"{FIVETRAN_API_BASE}/{FIVETRAN_API_VERSION_PATH}{FIVETRAN_CONNECTOR_PATH}{DEFAULT_CONNECTOR_ID}"
@@ -70,12 +71,15 @@ def test_fivetran_asset_run(tables, infer_missing_tables, should_error, schema_p
         poll_interval=0.1,
         poll_timeout=10,
         infer_missing_tables=infer_missing_tables,
+        op_tags=op_tags,
     )
 
     # expect the multi asset to have one asset key and one output for each specified asset key
     assert fivetran_assets[0].keys == {AssetKey(table.split(".")) for table in tables}
     assert len(fivetran_assets[0].op.output_defs) == len(tables)
 
+    assert fivetran_assets[0].op.tags == {**{"kind": "fivetran"}, **(op_tags or {})}
+
     fivetran_assets_job = build_assets_job(
         name="fivetran_assets_job",
         assets=fivetran_assets,