Merge branch 'master' of https://github.com/dagster-io/dagster into salazarm/status-selection-syntax

Author: salazarm

docs/docs/integrations/libraries/fivetran.md

@@ -127,6 +127,20 @@ Definitions from multiple Fivetran workspaces can be combined by instantiating m
   language="python"
 />
 
+### Define upstream dependencies
+
+By default, Dagster does not set upstream dependencies when generating asset specs for your Fivetran assets. You can set upstream dependencies on your Fivetran assets by passing an instance of the custom <PyObject section="libraries" module="dagster_fivetran" object="DagsterFivetranTranslator" /> to the <PyObject section="libraries" module="dagster_fivetran" object="load_fivetran_asset_specs" /> function.
+
+<CodeExample
+  startAfter="start_upstream_asset"
+  endBefore="end_upstream_asset"
+  path="docs_snippets/docs_snippets/integrations/fivetran/define_upstream_dependencies.py"
+/>
+
+Note that `super()` is called in each of the overridden methods to generate the default asset spec. It is best practice to generate the default asset spec before customizing it.
+
+You can pass an instance of the custom <PyObject section="libraries" module="dagster_fivetran" object="DagsterFivetranTranslator" /> to the <PyObject section="libraries" module="dagster_fivetran" object="fivetran_assets" /> decorator or the <PyObject section="libraries" module="dagster_fivetran" object="build_fivetran_assets_definitions" /> factory.
+
 ### Define downstream dependencies
 
 Dagster allows you to define assets that are downstream of specific Fivetran tables using their asset keys. The asset key for a Fivetran table can be retrieved using the asset definitions created using the <PyObject section="libraries" module="dagster_fivetran" object="fivetran_assets" /> decorator. The below example defines `my_downstream_asset` as a downstream dependency of `my_fivetran_table`:

examples/docs_snippets/docs_snippets/integrations/fivetran/define_upstream_dependencies.py

@@ -0,0 +1,32 @@
+from dagster_fivetran import (
+    DagsterFivetranTranslator,
+    FivetranConnectorTableProps,
+    FivetranWorkspace,
+    load_fivetran_asset_specs,
+)
+
+import dagster as dg
+
+fivetran_workspace = FivetranWorkspace(
+    account_id=dg.EnvVar("FIVETRAN_ACCOUNT_ID"),
+    api_key=dg.EnvVar("FIVETRAN_API_KEY"),
+    api_secret=dg.EnvVar("FIVETRAN_API_SECRET"),
+)
+
+
+# start_upstream_asset
+class MyCustomFivetranTranslator(DagsterFivetranTranslator):
+    def get_asset_spec(self, props: FivetranConnectorTableProps) -> dg.AssetSpec:
+        # We create the default asset spec using super()
+        default_spec = super().get_asset_spec(props)
+        # We set an upstream dependency for our assets
+        return default_spec.replace_attributes(deps=["my_upstream_asset_key"])
+
+
+fivetran_specs = load_fivetran_asset_specs(
+    fivetran_workspace, dagster_fivetran_translator=MyCustomFivetranTranslator()
+)
+
+# end_upstream_asset
+
+defs = dg.Definitions(assets=fivetran_specs, resources={"fivetran": fivetran_workspace})

examples/docs_snippets/docs_snippets_tests/test_integration_files_load.py

@@ -22,6 +22,7 @@
     f"{snippets_folder}/fivetran/customize_fivetran_asset_defs.py",
     f"{snippets_folder}/fivetran/customize_fivetran_translator_asset_spec.py",
     f"{snippets_folder}/fivetran/define_downstream_dependencies.py",
+    f"{snippets_folder}/fivetran/define_upstream_dependencies.py",
     f"{snippets_folder}/fivetran/multiple_fivetran_workspaces.py",
     f"{snippets_folder}/fivetran/representing_fivetran_assets.py",
     f"{snippets_folder}/fivetran/select_fivetran_connectors.py",

js_modules/dagster-ui/packages/ui-core/src/assets/LaunchAssetChoosePartitionsDialog.tsx

@@ -60,6 +60,7 @@ import {
 } from '../instance/backfill/types/BackfillUtils.types';
 import {fetchTagsAndConfigForAssetJob} from '../launchpad/ConfigFetch';
 import {TagContainer, TagEditor} from '../launchpad/TagEditor';
+import {tagsWithUIExecutionTags} from '../launchpad/uiExecutionTags';
 import {
   DAEMON_NOT_RUNNING_ALERT_INSTANCE_FRAGMENT,
   DaemonNotRunningAlert,
@@ -311,10 +312,11 @@ const LaunchAssetChoosePartitionsDialogBody = ({
   };
 
   const onLaunchAsBackfill = async () => {
+    const backfillTags = tagsWithUIExecutionTags(tags);
     const backfillParams: LaunchBackfillParams =
       target.type === 'job' && !isHiddenAssetGroupJob(target.jobName)
         ? {
-            tags,
+            tags: backfillTags,
             assetSelection: assets.map(asAssetKeyInput),
             partitionNames: keysFiltered,
             fromFailure: false,
@@ -329,12 +331,12 @@ const LaunchAssetChoosePartitionsDialogBody = ({
           }
         : target.type === 'pureAll'
           ? {
-              tags,
+              tags: backfillTags,
               assetSelection: assets.map(asAssetKeyInput),
               allPartitions: true,
             }
           : {
-              tags,
+              tags: backfillTags,
               assetSelection: assets.map(asAssetKeyInput),
               partitionNames: keysFiltered,
               fromFailure: false,

js_modules/dagster-ui/packages/ui-core/src/assets/__tests__/LaunchAssetExecutionButton.test.tsx

@@ -7,6 +7,7 @@ import {CustomAlertProvider} from '../../app/CustomAlertProvider';
 import {CustomConfirmationProvider} from '../../app/CustomConfirmationProvider';
 import {displayNameForAssetKey} from '../../asset-graph/Utils';
 import {LaunchPartitionBackfillMutation} from '../../instance/backfill/types/BackfillUtils.types';
+import {UI_EXECUTION_TAGS} from '../../launchpad/uiExecutionTags';
 import {LaunchPipelineExecutionMutation} from '../../runs/types/RunUtils.types';
 import {TestProvider} from '../../testing/TestProvider';
 import {buildWorkspaceMocks} from '../../workspace/WorkspaceContext/__fixtures__/Workspace.fixtures';
@@ -190,7 +191,7 @@ describe('LaunchAssetExecutionButton', () => {
       const launchMock = buildExpectedLaunchSingleRunMutation({
         mode: 'default',
         executionMetadata: {
-          tags: [],
+          tags: [...UI_EXECUTION_TAGS],
         },
         runConfigData: '{}',
         selector: {
@@ -214,7 +215,7 @@ describe('LaunchAssetExecutionButton', () => {
       it('should not include checks if the job in context is marked without_checks', async () => {
         const launchMock = buildExpectedLaunchSingleRunMutation({
           mode: 'default',
-          executionMetadata: {tags: []},
+          executionMetadata: {tags: [...UI_EXECUTION_TAGS]},
           runConfigData: '{}',
           selector: {
             repositoryLocationName: 'test.py',
@@ -236,7 +237,7 @@ describe('LaunchAssetExecutionButton', () => {
       it('should include checks if the job in context includes them', async () => {
         const launchMock = buildExpectedLaunchSingleRunMutation({
           mode: 'default',
-          executionMetadata: {tags: []},
+          executionMetadata: {tags: [...UI_EXECUTION_TAGS]},
           runConfigData: '{}',
           selector: {
             repositoryLocationName: 'test.py',
@@ -275,7 +276,7 @@ describe('LaunchAssetExecutionButton', () => {
       const launchMock = buildExpectedLaunchSingleRunMutation({
         mode: 'default',
         executionMetadata: {
-          tags: [],
+          tags: [...UI_EXECUTION_TAGS],
         },
         runConfigData: '{}',
         selector: {
@@ -327,6 +328,7 @@ describe('LaunchAssetExecutionButton', () => {
           tags: [
             {key: 'dagster/partition', value: '2023-02-22'},
             {key: 'dagster/partition_set', value: 'my_asset_job_partition_set'},
+            ...UI_EXECUTION_TAGS,
           ],
         },
         mode: 'default',
@@ -360,7 +362,7 @@ describe('LaunchAssetExecutionButton', () => {
         assetSelection: [{path: ['asset_daily']}],
         partitionNames: ASSET_DAILY_PARTITION_KEYS,
         fromFailure: false,
-        tags: [],
+        tags: [...UI_EXECUTION_TAGS],
       });
       renderButton({
         scope: {all: [ASSET_DAILY]},
@@ -400,7 +402,7 @@ describe('LaunchAssetExecutionButton', () => {
         assetSelection: [{path: ['asset_daily']}],
         partitionNames: ASSET_DAILY_PARTITION_KEYS_MISSING,
         fromFailure: false,
-        tags: [],
+        tags: [...UI_EXECUTION_TAGS],
       });
       renderButton({
         scope: {all: [ASSET_DAILY]},
@@ -437,6 +439,7 @@ describe('LaunchAssetExecutionButton', () => {
           tags: [
             {key: 'dagster/partition', value: '2023-02-22'},
             {key: 'dagster/partition_set', value: '__ASSET_JOB_partition_set'},
+            ...UI_EXECUTION_TAGS,
           ],
         },
         mode: 'default',
@@ -472,7 +475,7 @@ describe('LaunchAssetExecutionButton', () => {
           assetSelection: [{path: ['asset_daily']}],
           partitionNames: ASSET_DAILY_PARTITION_KEYS,
           fromFailure: false,
-          tags: [],
+          tags: [...UI_EXECUTION_TAGS],
         });
         renderButton({
           scope: {all: [ASSET_DAILY]},
@@ -503,6 +506,7 @@ describe('LaunchAssetExecutionButton', () => {
             tags: [
               {key: 'dagster/asset_partition_range_start', value: '2020-01-02'},
               {key: 'dagster/asset_partition_range_end', value: '2023-02-22'},
+              ...UI_EXECUTION_TAGS,
             ],
           },
           runConfigData: '{}\n',
@@ -540,6 +544,7 @@ describe('LaunchAssetExecutionButton', () => {
             tags: [
               {key: 'dagster/asset_partition_range_start', value: '2020-01-02'},
               {key: 'dagster/asset_partition_range_end', value: '2023-02-22'},
+              ...UI_EXECUTION_TAGS,
             ],
           },
           runConfigData: '{}\n',
@@ -572,7 +577,7 @@ describe('LaunchAssetExecutionButton', () => {
         assetSelection: [{path: ['asset_daily']}, {path: ['asset_weekly']}],
         partitionNames: ASSET_DAILY_PARTITION_KEYS,
         fromFailure: false,
-        tags: [],
+        tags: [...UI_EXECUTION_TAGS],
       });
 
       renderButton({
@@ -604,7 +609,7 @@ describe('LaunchAssetExecutionButton', () => {
         assetSelection: [{path: ['asset_daily']}, {path: ['asset_weekly']}],
         partitionNames: ASSET_DAILY_PARTITION_KEYS,
         fromFailure: false,
-        tags: [],
+        tags: [...UI_EXECUTION_TAGS],
       });
 
       renderButton({
@@ -624,7 +629,7 @@ describe('LaunchAssetExecutionButton', () => {
 
     it('should offer to materialize all partitions if roots have different partition defintions ("pureAll" case)', async () => {
       const LaunchPureAllMutationMock = buildExpectedLaunchBackfillMutation({
-        tags: [],
+        tags: [...UI_EXECUTION_TAGS],
         assetSelection: [
           {path: ['asset_daily']},
           {path: ['asset_weekly']},
@@ -658,7 +663,7 @@ describe('LaunchAssetExecutionButton', () => {
     it('should present a warning if pre-flight check indicates other asset keys are required', async () => {
       const launchMock = buildExpectedLaunchSingleRunMutation({
         mode: 'default',
-        executionMetadata: {tags: []},
+        executionMetadata: {tags: [...UI_EXECUTION_TAGS]},
         runConfigData: '{}',
         selector: {
           repositoryLocationName: 'test.py',

js_modules/dagster-ui/packages/ui-core/src/launchpad/uiExecutionTags.tsx

@@ -0,0 +1,21 @@
+import {ExecutionParams, ExecutionTag} from '../graphql/types';
+import {DagsterTag} from '../runs/RunTag';
+
+export const UI_EXECUTION_TAGS: ExecutionTag[] = [{key: DagsterTag.FromUI, value: 'true'}];
+
+export function tagsWithUIExecutionTags(tags: ExecutionTag[] | null | undefined) {
+  return [
+    ...(tags || []).filter((t) => !UI_EXECUTION_TAGS.some((a) => a.key === t.key)),
+    ...UI_EXECUTION_TAGS,
+  ];
+}
+
+export function paramsWithUIExecutionTags(params: ExecutionParams) {
+  return {
+    ...params,
+    executionMetadata: {
+      ...params.executionMetadata,
+      tags: tagsWithUIExecutionTags(params.executionMetadata?.tags),
+    },
+  };
+}

js_modules/dagster-ui/packages/ui-core/src/launchpad/useLaunchMultipleRunsWithTelemetry.ts

@@ -3,6 +3,7 @@ import {useHistory} from 'react-router-dom';
 
 import {showLaunchError} from './showLaunchError';
 import {useMutation} from '../apollo-client';
+import {paramsWithUIExecutionTags} from './uiExecutionTags';
 import {TelemetryAction, useTelemetryAction} from '../app/Telemetry';
 import {
   LAUNCH_MULTIPLE_RUNS_MUTATION,
@@ -49,7 +50,15 @@ export function useLaunchMultipleRunsWithTelemetry() {
           opSelection: undefined,
         };
 
-        const result = (await launchMultipleRuns({variables})).data?.launchMultipleRuns;
+        const finalized = {
+          ...variables,
+          executionParamsList: Array.isArray(variables.executionParamsList)
+            ? variables.executionParamsList.map(paramsWithUIExecutionTags)
+            : paramsWithUIExecutionTags(variables.executionParamsList),
+        };
+
+        const result = (await launchMultipleRuns({variables: finalized})).data?.launchMultipleRuns;
+
         if (result) {
           handleLaunchMultipleResult(result, history, {behavior});
           logTelemetry(

js_modules/dagster-ui/packages/ui-core/src/launchpad/useLaunchWithTelemetry.oss.ts

@@ -3,6 +3,7 @@ import {useHistory} from 'react-router-dom';
 
 import {showLaunchError} from './showLaunchError';
 import {useMutation} from '../apollo-client';
+import {paramsWithUIExecutionTags} from './uiExecutionTags';
 import {TelemetryAction, useTelemetryAction} from '../app/Telemetry';
 import {
   LAUNCH_PIPELINE_EXECUTION_MUTATION,
@@ -25,21 +26,23 @@ export function useLaunchWithTelemetry() {
   const history = useHistory();
 
   return useCallback(
-    async (variables: LaunchPipelineExecutionMutationVariables, behavior: LaunchBehavior) => {
-      const jobName =
-        variables.executionParams.selector.jobName ||
-        variables.executionParams.selector.pipelineName;
+    async (
+      {executionParams, ...rest}: LaunchPipelineExecutionMutationVariables,
+      behavior: LaunchBehavior,
+    ) => {
+      const jobName = executionParams.selector.jobName || executionParams.selector.pipelineName;
 
       if (!jobName) {
         return;
       }
 
       const metadata: {[key: string]: string | null | undefined} = {
         jobName,
-        opSelection: variables.executionParams.selector.solidSelection ? 'provided' : undefined,
+        opSelection: executionParams.selector.solidSelection ? 'provided' : undefined,
       };
 
-      const result = await launchPipelineExecution({variables});
+      const finalized = {executionParams: paramsWithUIExecutionTags(executionParams), ...rest};
+      const result = await launchPipelineExecution({variables: finalized});
       logTelemetry(TelemetryAction.LAUNCH_RUN, metadata);
       try {
         handleLaunchResult(jobName, result.data?.launchPipelineExecution, history, {behavior});

js_modules/dagster-ui/packages/ui-core/src/partitions/BackfillSelector.tsx

@@ -43,6 +43,7 @@ import {
 } from '../instance/backfill/types/BackfillUtils.types';
 import {LaunchButton} from '../launchpad/LaunchButton';
 import {TagContainer, TagEditor} from '../launchpad/TagEditor';
+import {tagsWithUIExecutionTags} from '../launchpad/uiExecutionTags';
 import {explodeCompositesInHandleGraph} from '../pipelines/CompositeSupport';
 import {GRAPH_EXPLORER_SOLID_HANDLE_FRAGMENT} from '../pipelines/GraphExplorer';
 import {GraphQueryInput} from '../ui/GraphQueryInput';
@@ -388,7 +389,7 @@ const LaunchBackfillButton = ({
           partitionNames,
           reexecutionSteps,
           fromFailure,
-          tags,
+          tags: tagsWithUIExecutionTags(tags),
         },
       },
     });

js_modules/dagster-ui/packages/ui-core/src/runs/ReexecutionDialog.tsx

@@ -23,6 +23,7 @@ import {
   LaunchPipelineReexecutionMutationVariables,
 } from './types/RunUtils.types';
 import {ExecutionTag, ReexecutionStrategy} from '../graphql/types';
+import {tagsWithUIExecutionTags} from '../launchpad/uiExecutionTags';
 
 export interface ReexecutionDialogProps {
   isOpen: boolean;
@@ -179,7 +180,7 @@ export const ReexecutionDialog = (props: ReexecutionDialogProps) => {
     dispatch({type: 'start'});
 
     const runList = Object.keys(state.frozenRuns);
-    const extraTags = extraTagsValidated.toSave.length ? extraTagsValidated.toSave : undefined;
+    const extraTags = tagsWithUIExecutionTags(extraTagsValidated.toSave);
 
     for (const runId of runList) {
       const {data} = await reexecute({

js_modules/dagster-ui/packages/ui-core/src/runs/RunTag.tsx

@@ -33,6 +33,7 @@ export enum DagsterTag {
   AssetEvaluationID = 'dagster/asset_evaluation_id',
   SnapshotID = 'dagster/snapshot_id', // This only exists on the client, not the server.
   ReportingUser = 'dagster/reporting_user',
+  FromUI = 'dagster/from_ui',
   User = 'user',
 
   // Hidden tags (using ".dagster" HIDDEN_TAG_PREFIX)