{banner}
- {!flagLegacyNav && !didDismissNavAlert ? (
- dagster-powerbi)
+ dagster-sigma)
+ dagster-tableau)
+ | (pipe)
A | B
+ OR; either condition is true; ex: A | B
& (ampersand)
A & B
+ AND; both conditions are true; ex: A & B
dagster/relation_identifier
+ dagster/table_name
AutoMaterializePolicys. However,
- this example provides a starting point for those wishing to customize this
- behavior beyond what the pre-built sensor supports.
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
- - - -
- -If we kick off a run of the `rebuild_customers_list` DAG in Airflow, we should see the corresponding asset materialize in Dagster. - -- - - -
- -_Note: When the code location loads, Dagster will query the Airflow REST API in order to build a representation of your DAGs. In order for Dagster to reflect changes to your DAGs, you will need to reload your code location._ - -- - - -
- -Kicking off a run of the DAG in Airflow, you should see the newly created assets materialize in Dagster as each task completes. - -_Note: There will be some delay between task completion and assets materializing in Dagster, managed by the sensor. This sensor runs every 30 seconds by default (you can reduce down to one second via the `minimum_interval_seconds` argument to `sensor`), so there will be some delay._ - -## Migrating Assets - -Once you have created corresponding definitions in Dagster to your Airflow tasks, you can proxy execution to Dagster on a per-task basis while Airflow is still controlling scheduling and orchestration. -Once a task has been proxied, Airflow will kick off materializations of corresponding Dagster assets in place of executing the business logic of that task. - -To begin proxying tasks in a DAG, first you will need a file to track proxying state. In your Airflow DAG directory, create a `proxied_state` folder, and in it create a yaml file with the same name as your DAG. The included example at [`airflow_dags/proxied_state`](./tutorial_example/airflow_dags/proxied_state) is used by `make airflow_run`, and can be used as a template for your own proxied state files. - -Given our example DAG `rebuild_customers_list` with three tasks, `load_raw_customers`, `run_dbt_model`, and `export_customers`, [`proxied_state/rebuild_customers_list.yaml`](./tutorial_example/airflow_dags/proxied_state/rebuild_customers_list.yaml) should look like the following: - -```yaml -# tutorial_example/airflow_dags/proxied_state/rebuild_customers_list.yaml -tasks: - - id: load_raw_customers - proxied: False - - id: build_dbt_models - proxied: False - - id: export_customers - proxied: False -``` - -Next, you will need to modify your Airflow DAG to make it aware of the proxied state. This is already done in the example DAG: - -```python -# tutorial_example/snippets/dags_truncated.py -# Dags file can be found at tutorial_example/airflow_dags/dags.py -from pathlib import Path - -from airflow import DAG -from dagster_airlift.in_airflow import proxying_to_dagster -from dagster_airlift.in_airflow.proxied_state import load_proxied_state_from_yaml - -dag = DAG("rebuild_customers_list", ...) - -... - -# Set this to True to begin the proxying process -PROXYING = False - -if PROXYING: - proxying_to_dagster( - global_vars=globals(), - proxied_state=load_proxied_state_from_yaml(Path(__file__).parent / "proxied_state"), - ) -``` - -Set `PROXYING` to `True` or eliminate the `if` statement. - -The DAG will now display its proxied state in the Airflow UI. (There is some latency as Airflow evaluates the Python file periodically.) - -- - - -
- -### Migrating individual tasks - -In order to proxy a task, you must do two things: - -1. First, ensure all associated assets are executable in Dagster by providing asset definitions in place of bare asset specs. -2. The `proxied: False` status in the `proxied_state` YAML folder must be adjusted to `proxied: True`. - -Any task marked as proxied will use the `DefaultProxyToDagsterOperator` when executed as part of the DAG. This operator will use the Dagster GraphQL API to initiate a Dagster run of the assets corresponding to the task. - -The proxied file acts as the source of truth for proxied state. The information is attached to the DAG and then accessed by Dagster via the REST API. - -A task which has been proxied can be easily toggled back to run in Airflow (for example, if a bug in implementation was encountered) simply by editing the file to `proxied: False`. - -#### Supporting custom authorization - -If your Dagster deployment lives behind a custom auth backend, you can customize the Airflow-to-Dagster proxying behavior to authenticate to your backend. -`proxying_to_dagster` can take a parameter `dagster_operator_klass`, which allows you to define a custom `BaseProxyToDagsterOperator` class. This allows you to -override how a session is created. Let's say for example, your Dagster installation requires an access key to be set whenever a request is made, and that access key is set in an Airflow `Variable` called `my_api_key`. -We can create a custom `BaseProxyToDagsterOperator` subclass which will retrieve that variable value and set it on the session, so that any requests to Dagster's graphql API -will be made using that api key. - -```python -# tutorial_example/snippets/custom_operator_examples/custom_proxy.py -from pathlib import Path - -import requests -from airflow import DAG -from airflow.utils.context import Context -from dagster_airlift.in_airflow import BaseProxyTaskToDagsterOperator, proxying_to_dagster -from dagster_airlift.in_airflow.proxied_state import load_proxied_state_from_yaml - - -class CustomProxyToDagsterOperator(BaseProxyTaskToDagsterOperator): - def get_dagster_session(self, context: Context) -> requests.Session: - if "var" not in context: - raise ValueError("No variables found in context") - api_key = context["var"]["value"].get("my_api_key") - session = requests.Session() - session.headers.update({"Authorization": f"Bearer {api_key}"}) - return session - - def get_dagster_url(self, context: Context) -> str: - return "https://dagster.example.com/" - - -dag = DAG( - dag_id="custom_proxy_example", -) - -# At the end of your dag file -proxying_to_dagster( - global_vars=globals(), - proxied_state=load_proxied_state_from_yaml(Path(__file__).parent / "proxied_state"), - build_from_task_fn=CustomProxyToDagsterOperator.build_from_task, -) -``` - -#### Dagster Plus Authorization - -You can use a customer proxy operator to establish a connection to a Dagster plus deployment. The below example proxies to Dagster Plus using organization name, deployment name, and user token set as -Airflow Variables. To set a Dagster+ user token, follow this guide: https://docs.dagster.io/dagster-plus/account/managing-user-agent-tokens#managing-user-tokens. - -```python -# tutorial_example/snippets/custom_operator_examples/plus_proxy_operator.py -import requests -from airflow.utils.context import Context -from dagster_airlift.in_airflow import BaseProxyTaskToDagsterOperator - - -class DagsterCloudProxyOperator(BaseProxyTaskToDagsterOperator): - def get_variable(self, context: Context, var_name: str) -> str: - if "var" not in context: - raise ValueError("No variables found in context") - return context["var"]["value"][var_name] - - def get_dagster_session(self, context: Context) -> requests.Session: - dagster_cloud_user_token = self.get_variable(context, "dagster_cloud_user_token") - session = requests.Session() - session.headers.update({"Dagster-Cloud-Api-Token": dagster_cloud_user_token}) - return session - - def get_dagster_url(self, context: Context) -> str: - org_name = self.get_variable(context, "dagster_plus_organization_name") - deployment_name = self.get_variable(context, "dagster_plus_deployment_name") - return f"https://{org_name}.dagster.plus/{deployment_name}" -``` - -#### Migrating common operators - -For some common operator patterns, like our dbt operator, Dagster supplies factories to build software defined assets for our tasks. In fact, the `@dbt_assets` decorator used earlier already backs its assets with definitions, so we can toggle the proxied state of the `build_dbt_models` task to `proxied: True` in the proxied state file: - -```yaml -# tutorial_example/snippets/dbt_proxied.yaml -tasks: - - id: load_raw_customers - proxied: False - - id: build_dbt_models - proxied: True - - id: export_customers - proxied: False -``` - -**Important**: It may take up to 30 seconds for the proxied state in the Airflow UI to reflect this change. You must subsequently reload the definitions in Dagster via the UI or by restarting `dagster dev`. - -You can now run the `rebuild_customers_list` DAG in Airflow, and the `build_dbt_models` task will be executed in a Dagster run: - -- - - -
- -You'll note that we proxied a task in the _middle_ of the Airflow DAG. The Airflow DAG structure and execution history is stable in the Airflow UI, but execution of `build_dbt_models` has moved to Dagster. - -#### Migrating the remaining custom operators - -For all other operator types, we will need to build our own asset definitions. We recommend creating a factory function whose arguments match the inputs to your Airflow operator. Then, you can use this factory to build definitions for each Airflow task. - -For example, our `load_raw_customers` task uses a custom `LoadCSVToDuckDB` operator. We'll define a function `load_csv_to_duckdb_defs` factory to build corresponding software-defined assets. Similarly for `export_customers` we'll define a function `export_duckdb_to_csv_defs` to build SDAs: - -```python -# tutorial_example/dagster_defs/stages/migrate.py -import os -from pathlib import Path - -from dagster import ( - AssetExecutionContext, - AssetsDefinition, - AssetSpec, - Definitions, - materialize, - multi_asset, -) -from dagster_airlift.core import ( - AirflowInstance, - BasicAuthBackend, - assets_with_task_mappings, - build_defs_from_airflow_instance, -) -from dagster_dbt import DbtCliResource, DbtProject, dbt_assets - -# Code also invoked from Airflow -from tutorial_example.shared.export_duckdb_to_csv import ExportDuckDbToCsvArgs, export_duckdb_to_csv -from tutorial_example.shared.load_csv_to_duckdb import LoadCsvToDuckDbArgs, load_csv_to_duckdb - - -def dbt_project_path() -> Path: - env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") - assert env_val, "TUTORIAL_DBT_PROJECT_DIR must be set" - return Path(env_val) - - -def airflow_dags_path() -> Path: - return Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "tutorial_example" / "airflow_dags" - - -def load_csv_to_duckdb_asset(spec: AssetSpec, args: LoadCsvToDuckDbArgs) -> AssetsDefinition: - @multi_asset(name=f"load_{args.table_name}", specs=[spec]) - def _multi_asset() -> None: - load_csv_to_duckdb(args) - - return _multi_asset - - -def export_duckdb_to_csv_defs(spec: AssetSpec, args: ExportDuckDbToCsvArgs) -> AssetsDefinition: - @multi_asset(name=f"export_{args.table_name}", specs=[spec]) - def _multi_asset() -> None: - export_duckdb_to_csv(args) - - return _multi_asset - - -@dbt_assets( - manifest=dbt_project_path() / "target" / "manifest.json", - project=DbtProject(dbt_project_path()), -) -def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): - yield from dbt.cli(["build"], context=context).stream() - - -mapped_assets = assets_with_task_mappings( - dag_id="rebuild_customers_list", - task_mappings={ - "load_raw_customers": [ - load_csv_to_duckdb_asset( - AssetSpec(key=["raw_data", "raw_customers"]), - LoadCsvToDuckDbArgs( - table_name="raw_customers", - csv_path=airflow_dags_path() / "raw_customers.csv", - duckdb_path=Path(os.environ["AIRFLOW_HOME"]) / "jaffle_shop.duckdb", - names=["id", "first_name", "last_name"], - duckdb_schema="raw_data", - duckdb_database_name="jaffle_shop", - ), - ) - ], - "build_dbt_models": - # load rich set of assets from dbt project - [dbt_project_assets], - "export_customers": [ - export_duckdb_to_csv_defs( - AssetSpec(key="customers_csv", deps=["customers"]), - ExportDuckDbToCsvArgs( - table_name="customers", - csv_path=Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv", - duckdb_path=Path(os.environ["AIRFLOW_HOME"]) / "jaffle_shop.duckdb", - duckdb_database_name="jaffle_shop", - ), - ) - ], - }, -) - - -defs = build_defs_from_airflow_instance( - airflow_instance=AirflowInstance( - auth_backend=BasicAuthBackend( - webserver_url="http://localhost:8080", - username="admin", - password="admin", - ), - name="airflow_instance_one", - ), - defs=Definitions( - assets=mapped_assets, - resources={"dbt": DbtCliResource(project_dir=dbt_project_path())}, - ), -) -``` - -We can then toggle the proxied state of the remaining tasks in the `proxied_state` file: - -```yaml -# tutorial_example/snippets/all_proxied.yaml -tasks: - - id: load_raw_customers - proxied: True - - id: build_dbt_models - proxied: True - - id: export_customers - proxied: True -``` - -## Decomissioning an Airflow DAG - -Once we are confident in our migrated versions of the tasks, we can decommission the Airflow DAG. First, we can remove the DAG from our Airflow DAG directory. - -Next, we can strip the task associations from our Dagster definitions. This can be done by removing the `assets_with_task_mappings` call. We can use this opportunity to attach our assets to a `ScheduleDefinition` so that Dagster's scheduler can manage their execution: - -```python -# tutorial_example/dagster_defs/stages/standalone.py -import os -from pathlib import Path - -from dagster import ( - AssetCheckResult, - AssetCheckSeverity, - AssetExecutionContext, - AssetKey, - AssetsDefinition, - AssetSelection, - AssetSpec, - Definitions, - ScheduleDefinition, - asset_check, - multi_asset, -) -from dagster_dbt import DbtCliResource, DbtProject, dbt_assets - -# Code also invoked from Airflow -from tutorial_example.shared.export_duckdb_to_csv import ExportDuckDbToCsvArgs, export_duckdb_to_csv -from tutorial_example.shared.load_csv_to_duckdb import LoadCsvToDuckDbArgs, load_csv_to_duckdb - - -def dbt_project_path() -> Path: - env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") - assert env_val, "TUTORIAL_DBT_PROJECT_DIR must be set" - return Path(env_val) - - -def airflow_dags_path() -> Path: - return Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "tutorial_example" / "airflow_dags" - - -def load_csv_to_duckdb_asset(spec: AssetSpec, args: LoadCsvToDuckDbArgs) -> AssetsDefinition: - @multi_asset(name=f"load_{args.table_name}", specs=[spec]) - def _multi_asset() -> None: - load_csv_to_duckdb(args) - - return _multi_asset - - -def export_duckdb_to_csv_defs(spec: AssetSpec, args: ExportDuckDbToCsvArgs) -> AssetsDefinition: - @multi_asset(name=f"export_{args.table_name}", specs=[spec]) - def _multi_asset() -> None: - export_duckdb_to_csv(args) - - return _multi_asset - - -@dbt_assets( - manifest=dbt_project_path() / "target" / "manifest.json", - project=DbtProject(dbt_project_path()), -) -def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): - yield from dbt.cli(["build"], context=context).stream() - - -assets = [ - load_csv_to_duckdb_asset( - AssetSpec(key=["raw_data", "raw_customers"]), - LoadCsvToDuckDbArgs( - table_name="raw_customers", - csv_path=airflow_dags_path() / "raw_customers.csv", - duckdb_path=Path(os.environ["AIRFLOW_HOME"]) / "jaffle_shop.duckdb", - names=["id", "first_name", "last_name"], - duckdb_schema="raw_data", - duckdb_database_name="jaffle_shop", - ), - ), - dbt_project_assets, - export_duckdb_to_csv_defs( - AssetSpec(key="customers_csv", deps=["customers"]), - ExportDuckDbToCsvArgs( - table_name="customers", - csv_path=Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv", - duckdb_path=Path(os.environ["AIRFLOW_HOME"]) / "jaffle_shop.duckdb", - duckdb_database_name="jaffle_shop", - ), - ), -] - - -@asset_check(asset=AssetKey(["customers_csv"])) -def validate_exported_csv() -> AssetCheckResult: - csv_path = Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv" - - if not csv_path.exists(): - return AssetCheckResult(passed=False, description=f"Export CSV {csv_path} does not exist") - - rows = len(csv_path.read_text().split("\n")) - if rows < 2: - return AssetCheckResult( - passed=False, - description=f"Export CSV {csv_path} is empty", - severity=AssetCheckSeverity.WARN, - ) - - return AssetCheckResult( - passed=True, - description=f"Export CSV {csv_path} exists", - metadata={"rows": rows}, - ) - - -rebuild_customer_list_schedule = rebuild_customers_list_schedule = ScheduleDefinition( - name="rebuild_customers_list_schedule", - target=AssetSelection.assets(*assets), - cron_schedule="0 0 * * *", -) - - -defs = Definitions( - assets=assets, - schedules=[rebuild_customer_list_schedule], - asset_checks=[validate_exported_csv], - resources={"dbt": DbtCliResource(project_dir=dbt_project_path())}, -) -``` - -## Addendum: Adding asset checks - -Once you have peered your Airflow DAGs in Dagster, regardless of migration progress, you can begin to add asset checks to your Dagster code. Asset checks can be used to validate the quality of your data assets, and can provide additional observability and value on top of your Airflow DAG even before migration starts. - -For example, given a peered version of our DAG, we can add an asset check to ensure that the final `customers` CSV output exists and has a non-zero number of rows: - -```python -# tutorial_example/dagster_defs/stages/peer_with_check.py -import os -from pathlib import Path - -from dagster import AssetCheckResult, AssetCheckSeverity, AssetKey, Definitions, asset_check -from dagster_airlift.core import AirflowInstance, BasicAuthBackend, build_defs_from_airflow_instance - - -# Attach a check to the DAG representation asset, which will be executed by Dagster -# any time the DAG is run in Airflow -@asset_check(asset=AssetKey(["airflow_instance_one", "dag", "rebuild_customers_list"])) -def validate_exported_csv() -> AssetCheckResult: - csv_path = Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv" - - if not csv_path.exists(): - return AssetCheckResult(passed=False, description=f"Export CSV {csv_path} does not exist") - - rows = len(csv_path.read_text().split("\n")) - if rows < 2: - return AssetCheckResult( - passed=False, - description=f"Export CSV {csv_path} is empty", - severity=AssetCheckSeverity.WARN, - ) - - return AssetCheckResult( - passed=True, - description=f"Export CSV {csv_path} exists", - metadata={"rows": rows}, - ) - - -defs = Definitions.merge( - build_defs_from_airflow_instance( - airflow_instance=AirflowInstance( - # other backends available (e.g. MwaaSessionAuthBackend) - auth_backend=BasicAuthBackend( - webserver_url="http://localhost:8080", - username="admin", - password="admin", - ), - name="airflow_instance_one", - ) - ), - Definitions(asset_checks=[validate_exported_csv]), -) -``` - -Once we have introduced representations of the assets produced by our Airflow tasks, we can directly attach asset checks to these assets. These checks will run once the corresponding task completes, regardless of whether the task is executed in Airflow or Dagster. - -- - - - -
- -When performing dag-level mapping, we don't preserve task structure in the Airflow dags. This single task will materialize all mapped Dagster assets instead of executing the original Airflow task business logic. - -We can similarly mark `proxied` back to `False`, and the original task structure and business logic will return unchanged. - -### Customizing DAG proxying operator - -Similar to how we can customize the operator we construct on a per-dag basis, we can customize the operator we construct on a per-dag basis. We can use the `build_from_dag_fn` argument of `proxying_to_dagster` to provide a custom operator in place of the default. - -For example, let's take a look at the following custom operator which expects an API key to be provided as a variable: - -```python -# tutorial_example/snippets/custom_operator_examples/custom_dag_level_proxy.py -from pathlib import Path - -import requests -from airflow import DAG -from airflow.utils.context import Context -from dagster_airlift.in_airflow import BaseProxyDAGToDagsterOperator, proxying_to_dagster -from dagster_airlift.in_airflow.proxied_state import load_proxied_state_from_yaml - - -class CustomProxyToDagsterOperator(BaseProxyDAGToDagsterOperator): - def get_dagster_session(self, context: Context) -> requests.Session: - if "var" not in context: - raise ValueError("No variables found in context") - api_key = context["var"]["value"].get("my_api_key") - session = requests.Session() - session.headers.update({"Authorization": f"Bearer {api_key}"}) - return session - - def get_dagster_url(self, context: Context) -> str: - return "https://dagster.example.com/" - - # This method controls how the operator is built from the dag. - @classmethod - def build_from_dag(cls, dag: DAG): - return CustomProxyToDagsterOperator(dag=dag, task_id="OVERRIDDEN") - - -dag = DAG( - dag_id="custom_dag_level_proxy_example", -) - -# At the end of your dag file -proxying_to_dagster( - global_vars=globals(), - proxied_state=load_proxied_state_from_yaml(Path(__file__).parent / "proxied_state"), - build_from_dag_fn=CustomProxyToDagsterOperator.build_from_dag, -) -``` - -`BaseProxyDAGToDagsterOperator` has three abstract methods which must be implemented: - -- `get_dagster_session`, which controls the creation of a valid session to access the Dagster graphql API. -- `get_dagster_url`, which retrieves the domain at which the dagster webserver lives. -- `build_from_dag`, which controls how the proxying task is constructed from the provided DAG. diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/images/partitioned_mat.png b/examples/experimental/dagster-airlift/examples/tutorial-example/images/partitioned_mat.png new file mode 100644 index 0000000000000..2a50ca8eec0e3 Binary files /dev/null and b/examples/experimental/dagster-airlift/examples/tutorial-example/images/partitioned_mat.png differ diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/images/proxied_dag.png b/examples/experimental/dagster-airlift/examples/tutorial-example/images/proxied_dag.png index dbc7fbbc898e6..b50af91c37b34 100644 Binary files a/examples/experimental/dagster-airlift/examples/tutorial-example/images/proxied_dag.png and b/examples/experimental/dagster-airlift/examples/tutorial-example/images/proxied_dag.png differ diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/images/state_in_airflow.png b/examples/experimental/dagster-airlift/examples/tutorial-example/images/state_in_airflow.png index 1fa81a964c77d..d90301fc59788 100644 Binary files a/examples/experimental/dagster-airlift/examples/tutorial-example/images/state_in_airflow.png and b/examples/experimental/dagster-airlift/examples/tutorial-example/images/state_in_airflow.png differ diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/airflow_dags/dags.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/airflow_dags/dags.py index 975d21dec26a7..049bc5cc30c3a 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/airflow_dags/dags.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/airflow_dags/dags.py @@ -1,12 +1,13 @@ # Define the default arguments for the DAG import os -from datetime import datetime, timedelta +from datetime import timedelta from pathlib import Path from typing import List, Optional from airflow import DAG from airflow.models.operator import BaseOperator from airflow.operators.bash import BashOperator +from dagster._time import get_current_datetime_midnight from dagster_airlift.in_airflow import proxying_to_dagster from dagster_airlift.in_airflow.proxied_state import load_proxied_state_from_yaml from tutorial_example.shared.export_duckdb_to_csv import ExportDuckDbToCsvArgs, export_duckdb_to_csv @@ -78,7 +79,7 @@ def execute(self, context) -> None: default_args = { "owner": "airflow", "depends_on_past": False, - "start_date": datetime(2024, 7, 18), + "start_date": get_current_datetime_midnight(), "retries": 1, "retry_delay": timedelta(minutes=5), } @@ -87,7 +88,7 @@ def execute(self, context) -> None: dag = DAG( "rebuild_customers_list", default_args=default_args, - schedule_interval=None, + schedule="@daily", is_paused_upon_creation=False, ) diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate.py index 61e3f7c8f9f5a..4e77cbcac686d 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate.py @@ -5,13 +5,15 @@ AssetExecutionContext, AssetsDefinition, AssetSpec, + DailyPartitionsDefinition, Definitions, materialize, multi_asset, ) +from dagster._time import get_current_datetime_midnight from dagster_airlift.core import ( + AirflowBasicAuthBackend, AirflowInstance, - BasicAuthBackend, assets_with_task_mappings, build_defs_from_airflow_instance, ) @@ -21,6 +23,8 @@ from tutorial_example.shared.export_duckdb_to_csv import ExportDuckDbToCsvArgs, export_duckdb_to_csv from tutorial_example.shared.load_csv_to_duckdb import LoadCsvToDuckDbArgs, load_csv_to_duckdb +PARTITIONS_DEF = DailyPartitionsDefinition(start_date=get_current_datetime_midnight()) + def dbt_project_path() -> Path: env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") @@ -51,6 +55,7 @@ def _multi_asset() -> None: @dbt_assets( manifest=dbt_project_path() / "target" / "manifest.json", project=DbtProject(dbt_project_path()), + partitions_def=PARTITIONS_DEF, ) def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): yield from dbt.cli(["build"], context=context).stream() @@ -61,7 +66,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): task_mappings={ "load_raw_customers": [ load_csv_to_duckdb_asset( - AssetSpec(key=["raw_data", "raw_customers"]), + AssetSpec(key=["raw_data", "raw_customers"], partitions_def=PARTITIONS_DEF), LoadCsvToDuckDbArgs( table_name="raw_customers", csv_path=airflow_dags_path() / "raw_customers.csv", @@ -77,7 +82,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): [dbt_project_assets], "export_customers": [ export_duckdb_to_csv_defs( - AssetSpec(key="customers_csv", deps=["customers"]), + AssetSpec(key="customers_csv", deps=["customers"], partitions_def=PARTITIONS_DEF), ExportDuckDbToCsvArgs( table_name="customers", csv_path=Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv", @@ -92,7 +97,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): defs = build_defs_from_airflow_instance( airflow_instance=AirflowInstance( - auth_backend=BasicAuthBackend( + auth_backend=AirflowBasicAuthBackend( webserver_url="http://localhost:8080", username="admin", password="admin", diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_dag_level.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_dag_level.py index 9c49f4624cb96..cfebc83ad54eb 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_dag_level.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_dag_level.py @@ -10,8 +10,8 @@ multi_asset, ) from dagster_airlift.core import ( + AirflowBasicAuthBackend, AirflowInstance, - BasicAuthBackend, assets_with_dag_mappings, build_defs_from_airflow_instance, ) @@ -87,7 +87,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): defs = build_defs_from_airflow_instance( airflow_instance=AirflowInstance( - auth_backend=BasicAuthBackend( + auth_backend=AirflowBasicAuthBackend( webserver_url="http://localhost:8080", username="admin", password="admin", diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_with_check.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_with_check.py index dbd7647307397..314327c7f11d7 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_with_check.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/migrate_with_check.py @@ -8,14 +8,16 @@ AssetKey, AssetsDefinition, AssetSpec, + DailyPartitionsDefinition, Definitions, asset_check, materialize, multi_asset, ) +from dagster._time import get_current_datetime_midnight from dagster_airlift.core import ( + AirflowBasicAuthBackend, AirflowInstance, - BasicAuthBackend, assets_with_task_mappings, build_defs_from_airflow_instance, ) @@ -25,6 +27,8 @@ from tutorial_example.shared.export_duckdb_to_csv import ExportDuckDbToCsvArgs, export_duckdb_to_csv from tutorial_example.shared.load_csv_to_duckdb import LoadCsvToDuckDbArgs, load_csv_to_duckdb +PARTITIONS_DEF = DailyPartitionsDefinition(start_date=get_current_datetime_midnight()) + def dbt_project_path() -> Path: env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") @@ -55,6 +59,7 @@ def _multi_asset() -> None: @dbt_assets( manifest=dbt_project_path() / "target" / "manifest.json", project=DbtProject(dbt_project_path()), + partitions_def=PARTITIONS_DEF, ) def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): yield from dbt.cli(["build"], context=context).stream() @@ -65,7 +70,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): task_mappings={ "load_raw_customers": [ load_csv_to_duckdb_asset( - AssetSpec(key=["raw_data", "raw_customers"]), + AssetSpec(key=["raw_data", "raw_customers"], partitions_def=PARTITIONS_DEF), LoadCsvToDuckDbArgs( table_name="raw_customers", csv_path=airflow_dags_path() / "raw_customers.csv", @@ -81,7 +86,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): [dbt_project_assets], "export_customers": [ export_duckdb_to_csv_defs( - AssetSpec(key="customers_csv", deps=["customers"]), + AssetSpec(key="customers_csv", deps=["customers"], partitions_def=PARTITIONS_DEF), ExportDuckDbToCsvArgs( table_name="customers", csv_path=Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv", @@ -118,7 +123,7 @@ def validate_exported_csv() -> AssetCheckResult: defs = build_defs_from_airflow_instance( airflow_instance=AirflowInstance( - auth_backend=BasicAuthBackend( + auth_backend=AirflowBasicAuthBackend( webserver_url="http://localhost:8080", username="admin", password="admin", diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe.py index 0dd594aeed3e2..73af301732ee8 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe.py @@ -1,16 +1,46 @@ import os from pathlib import Path -from dagster import AssetExecutionContext, AssetSpec, Definitions +from dagster import ( + AssetCheckResult, + AssetCheckSeverity, + AssetExecutionContext, + AssetKey, + AssetSpec, + Definitions, + asset_check, +) from dagster_airlift.core import ( + AirflowBasicAuthBackend, AirflowInstance, - BasicAuthBackend, assets_with_task_mappings, build_defs_from_airflow_instance, ) from dagster_dbt import DbtCliResource, DbtProject, dbt_assets +@asset_check(asset=AssetKey(["airflow_instance_one", "dag", "rebuild_customers_list"])) +def validate_exported_csv() -> AssetCheckResult: + csv_path = Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv" + + if not csv_path.exists(): + return AssetCheckResult(passed=False, description=f"Export CSV {csv_path} does not exist") + + rows = len(csv_path.read_text().split("\n")) + if rows < 2: + return AssetCheckResult( + passed=False, + description=f"Export CSV {csv_path} is empty", + severity=AssetCheckSeverity.WARN, + ) + + return AssetCheckResult( + passed=True, + description=f"Export CSV {csv_path} exists", + metadata={"rows": rows}, + ) + + def dbt_project_path() -> Path: env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") assert env_val, "TUTORIAL_DBT_PROJECT_DIR must be set" @@ -37,7 +67,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): defs = build_defs_from_airflow_instance( airflow_instance=AirflowInstance( - auth_backend=BasicAuthBackend( + auth_backend=AirflowBasicAuthBackend( webserver_url="http://localhost:8080", username="admin", password="admin", @@ -47,5 +77,6 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): defs=Definitions( assets=mapped_assets, resources={"dbt": DbtCliResource(project_dir=dbt_project_path())}, + asset_checks=[validate_exported_csv], ), ) diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_check_on_asset.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_check_on_asset.py new file mode 100644 index 0000000000000..360950591cf82 --- /dev/null +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_check_on_asset.py @@ -0,0 +1,82 @@ +import os +from pathlib import Path + +from dagster import ( + AssetCheckResult, + AssetCheckSeverity, + AssetExecutionContext, + AssetKey, + AssetSpec, + Definitions, + asset_check, +) +from dagster_airlift.core import ( + AirflowBasicAuthBackend, + AirflowInstance, + assets_with_task_mappings, + build_defs_from_airflow_instance, +) +from dagster_dbt import DbtCliResource, DbtProject, dbt_assets + + +@asset_check(asset=AssetKey(["customers_csv"])) +def validate_exported_csv() -> AssetCheckResult: + csv_path = Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv" + + if not csv_path.exists(): + return AssetCheckResult(passed=False, description=f"Export CSV {csv_path} does not exist") + + rows = len(csv_path.read_text().split("\n")) + if rows < 2: + return AssetCheckResult( + passed=False, + description=f"Export CSV {csv_path} is empty", + severity=AssetCheckSeverity.WARN, + ) + + return AssetCheckResult( + passed=True, + description=f"Export CSV {csv_path} exists", + metadata={"rows": rows}, + ) + + +def dbt_project_path() -> Path: + env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") + assert env_val, "TUTORIAL_DBT_PROJECT_DIR must be set" + return Path(env_val) + + +@dbt_assets( + manifest=dbt_project_path() / "target" / "manifest.json", + project=DbtProject(dbt_project_path()), +) +def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): + yield from dbt.cli(["build"], context=context).stream() + + +mapped_assets = assets_with_task_mappings( + dag_id="rebuild_customers_list", + task_mappings={ + "load_raw_customers": [AssetSpec(key=["raw_data", "raw_customers"])], + "build_dbt_models": [dbt_project_assets], + "export_customers": [AssetSpec(key="customers_csv", deps=["customers"])], + }, +) + + +defs = build_defs_from_airflow_instance( + airflow_instance=AirflowInstance( + auth_backend=AirflowBasicAuthBackend( + webserver_url="http://localhost:8080", + username="admin", + password="admin", + ), + name="airflow_instance_one", + ), + defs=Definitions( + assets=mapped_assets, + resources={"dbt": DbtCliResource(project_dir=dbt_project_path())}, + asset_checks=[validate_exported_csv], + ), +) diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_dag_level.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_dag_level.py index 9e9f077137454..72d2516ab0c61 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_dag_level.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_dag_level.py @@ -3,8 +3,8 @@ from dagster import AssetExecutionContext, AssetSpec, Definitions from dagster_airlift.core import ( + AirflowBasicAuthBackend, AirflowInstance, - BasicAuthBackend, assets_with_dag_mappings, build_defs_from_airflow_instance, ) @@ -39,7 +39,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): defs = build_defs_from_airflow_instance( airflow_instance=AirflowInstance( - auth_backend=BasicAuthBackend( + auth_backend=AirflowBasicAuthBackend( webserver_url="http://localhost:8080", username="admin", password="admin", diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_with_partitions.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_with_partitions.py new file mode 100644 index 0000000000000..a9d88160b09a1 --- /dev/null +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/observe_with_partitions.py @@ -0,0 +1,91 @@ +import os +from pathlib import Path + +from dagster import ( + AssetCheckResult, + AssetCheckSeverity, + AssetExecutionContext, + AssetKey, + AssetSpec, + DailyPartitionsDefinition, + Definitions, + asset_check, +) +from dagster._time import get_current_datetime_midnight +from dagster_airlift.core import ( + AirflowBasicAuthBackend, + AirflowInstance, + assets_with_task_mappings, + build_defs_from_airflow_instance, +) +from dagster_dbt import DbtCliResource, DbtProject, dbt_assets + +PARTITIONS_DEF = DailyPartitionsDefinition(start_date=get_current_datetime_midnight()) + + +@asset_check(asset=AssetKey(["customers_csv"])) +def validate_exported_csv() -> AssetCheckResult: + csv_path = Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv" + + if not csv_path.exists(): + return AssetCheckResult(passed=False, description=f"Export CSV {csv_path} does not exist") + + rows = len(csv_path.read_text().split("\n")) + if rows < 2: + return AssetCheckResult( + passed=False, + description=f"Export CSV {csv_path} is empty", + severity=AssetCheckSeverity.WARN, + ) + + return AssetCheckResult( + passed=True, + description=f"Export CSV {csv_path} exists", + metadata={"rows": rows}, + ) + + +def dbt_project_path() -> Path: + env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") + assert env_val, "TUTORIAL_DBT_PROJECT_DIR must be set" + return Path(env_val) + + +@dbt_assets( + manifest=dbt_project_path() / "target" / "manifest.json", + project=DbtProject(dbt_project_path()), + partitions_def=PARTITIONS_DEF, +) +def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): + yield from dbt.cli(["build"], context=context).stream() + + +mapped_assets = assets_with_task_mappings( + dag_id="rebuild_customers_list", + task_mappings={ + "load_raw_customers": [ + AssetSpec(key=["raw_data", "raw_customers"], partitions_def=PARTITIONS_DEF) + ], + "build_dbt_models": [dbt_project_assets], + "export_customers": [ + AssetSpec(key="customers_csv", deps=["customers"], partitions_def=PARTITIONS_DEF) + ], + }, +) + + +defs = build_defs_from_airflow_instance( + airflow_instance=AirflowInstance( + auth_backend=AirflowBasicAuthBackend( + webserver_url="http://localhost:8080", + username="admin", + password="admin", + ), + name="airflow_instance_one", + ), + defs=Definitions( + assets=mapped_assets, + resources={"dbt": DbtCliResource(project_dir=dbt_project_path())}, + asset_checks=[validate_exported_csv], + ), +) diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer.py index 83e080db5a66f..ffd908784ce80 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer.py @@ -1,9 +1,13 @@ -from dagster_airlift.core import AirflowInstance, BasicAuthBackend, build_defs_from_airflow_instance +from dagster_airlift.core import ( + AirflowBasicAuthBackend, + AirflowInstance, + build_defs_from_airflow_instance, +) defs = build_defs_from_airflow_instance( airflow_instance=AirflowInstance( # other backends available (e.g. MwaaSessionAuthBackend) - auth_backend=BasicAuthBackend( + auth_backend=AirflowBasicAuthBackend( webserver_url="http://localhost:8080", username="admin", password="admin", diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer_with_check.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer_with_check.py index 65f45e57bb642..6abcef989641d 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer_with_check.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/peer_with_check.py @@ -2,7 +2,11 @@ from pathlib import Path from dagster import AssetCheckResult, AssetCheckSeverity, AssetKey, Definitions, asset_check -from dagster_airlift.core import AirflowInstance, BasicAuthBackend, build_defs_from_airflow_instance +from dagster_airlift.core import ( + AirflowBasicAuthBackend, + AirflowInstance, + build_defs_from_airflow_instance, +) # Attach a check to the DAG representation asset, which will be executed by Dagster @@ -29,17 +33,15 @@ def validate_exported_csv() -> AssetCheckResult: ) -defs = Definitions.merge( - build_defs_from_airflow_instance( - airflow_instance=AirflowInstance( - # other backends available (e.g. MwaaSessionAuthBackend) - auth_backend=BasicAuthBackend( - webserver_url="http://localhost:8080", - username="admin", - password="admin", - ), - name="airflow_instance_one", - ) +defs = build_defs_from_airflow_instance( + airflow_instance=AirflowInstance( + # other backends available (e.g. MwaaSessionAuthBackend) + auth_backend=AirflowBasicAuthBackend( + webserver_url="http://localhost:8080", + username="admin", + password="admin", + ), + name="airflow_instance_one", ), - Definitions(asset_checks=[validate_exported_csv]), + defs=Definitions(asset_checks=[validate_exported_csv]), ) diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/standalone.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/standalone.py index 99340a7040e62..2eedee5788956 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/standalone.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example/dagster_defs/stages/standalone.py @@ -9,17 +9,21 @@ AssetsDefinition, AssetSelection, AssetSpec, + DailyPartitionsDefinition, Definitions, ScheduleDefinition, asset_check, multi_asset, ) +from dagster._time import get_current_datetime_midnight from dagster_dbt import DbtCliResource, DbtProject, dbt_assets # Code also invoked from Airflow from tutorial_example.shared.export_duckdb_to_csv import ExportDuckDbToCsvArgs, export_duckdb_to_csv from tutorial_example.shared.load_csv_to_duckdb import LoadCsvToDuckDbArgs, load_csv_to_duckdb +PARTITIONS_DEF = DailyPartitionsDefinition(start_date=get_current_datetime_midnight()) + def dbt_project_path() -> Path: env_val = os.getenv("TUTORIAL_DBT_PROJECT_DIR") @@ -50,6 +54,7 @@ def _multi_asset() -> None: @dbt_assets( manifest=dbt_project_path() / "target" / "manifest.json", project=DbtProject(dbt_project_path()), + partitions_def=PARTITIONS_DEF, ) def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): yield from dbt.cli(["build"], context=context).stream() @@ -57,7 +62,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): assets = [ load_csv_to_duckdb_asset( - AssetSpec(key=["raw_data", "raw_customers"]), + AssetSpec(key=["raw_data", "raw_customers"], partitions_def=PARTITIONS_DEF), LoadCsvToDuckDbArgs( table_name="raw_customers", csv_path=airflow_dags_path() / "raw_customers.csv", @@ -69,7 +74,7 @@ def dbt_project_assets(context: AssetExecutionContext, dbt: DbtCliResource): ), dbt_project_assets, export_duckdb_to_csv_defs( - AssetSpec(key="customers_csv", deps=["customers"]), + AssetSpec(key="customers_csv", deps=["customers"], partitions_def=PARTITIONS_DEF), ExportDuckDbToCsvArgs( table_name="customers", csv_path=Path(os.environ["TUTORIAL_EXAMPLE_DIR"]) / "customers.csv", diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/integration_tests/test_migrating_e2e.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/integration_tests/test_migrating_e2e.py index 81eb94a9064dd..9105f494529a2 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/integration_tests/test_migrating_e2e.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/integration_tests/test_migrating_e2e.py @@ -5,7 +5,7 @@ import pytest from dagster import DagsterInstance from dagster._core.storage.dagster_run import DagsterRunStatus -from dagster_airlift.core import AirflowInstance, BasicAuthBackend +from dagster_airlift.core import AirflowBasicAuthBackend, AirflowInstance from .utils import ( poll_for_materialization, @@ -23,7 +23,7 @@ def test_migration_status( both the Airflow DAGs and the Dagster asset definitions. """ instance = AirflowInstance( - auth_backend=BasicAuthBackend( + auth_backend=AirflowBasicAuthBackend( webserver_url="http://localhost:8080", username="admin", password="admin", diff --git a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/unit_tests/test_proxy_operators.py b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/unit_tests/test_proxy_operators.py index dcee6accbe53b..2900309914e39 100644 --- a/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/unit_tests/test_proxy_operators.py +++ b/examples/experimental/dagster-airlift/examples/tutorial-example/tutorial_example_tests/unit_tests/test_proxy_operators.py @@ -1,4 +1,4 @@ -def test_dagster_cloud_proxy_operator(mock_airflow_variable: None) -> None: +def test_dagster_cloud_proxy_operator() -> None: from tutorial_example.snippets.custom_operator_examples.plus_proxy_operator import ( DagsterCloudProxyOperator, ) @@ -26,7 +26,7 @@ def test_dagster_cloud_proxy_operator(mock_airflow_variable: None) -> None: ) -def test_custom_proxy_operator(mock_airflow_variable: None) -> None: +def test_custom_proxy_operator() -> None: from tutorial_example.snippets.custom_operator_examples.custom_proxy import ( CustomProxyToDagsterOperator, ) @@ -40,7 +40,7 @@ def test_custom_proxy_operator(mock_airflow_variable: None) -> None: assert session.headers["Authorization"] == "Bearer test_key" -def test_dag_override_operator(mock_airflow_variable: None) -> None: +def test_dag_override_operator() -> None: from tutorial_example.snippets.custom_operator_examples.custom_dag_level_proxy import ( CustomProxyToDagsterOperator, ) diff --git a/examples/experimental/dagster-airlift/scripts/airflow_setup.sh b/examples/experimental/dagster-airlift/scripts/airflow_setup.sh index 8d6bc4422898c..bb61e59c82bf9 100755 --- a/examples/experimental/dagster-airlift/scripts/airflow_setup.sh +++ b/examples/experimental/dagster-airlift/scripts/airflow_setup.sh @@ -1,37 +1,41 @@ #!/bin/bash -# Check if the path argument is provided -if [ -z "$1" ]; then - echo "Usage: $0{asset.partitionDefinition.description}
-