Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[dagster-dbt] update usage of deprecated context methods #19806

Merged
merged 2 commits into from
Feb 15, 2024
Merged

[dagster-dbt] update usage of deprecated context methods #19806

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7cf51ab
use underlying op_execution_context to access deprecated methods
jamiedemaria Feb 14, 2024
bc6f559
pr comments
jamiedemaria Feb 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
use underlying op_execution_context to access deprecated methods
  • Loading branch information
@jamiedemaria
jamiedemaria committed Feb 14, 2024
commit 7cf51abb97b48a534cc61ec81ea7a077c5233faf
2 changes: 1 addition & 1 deletion python_modules/libraries/dagster-dbt/dagster_dbt/cloud/asset_defs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -536,7 +536,7 @@ def _assets(context: AssetExecutionContext):
# Run the dbt Cloud job to rematerialize the assets.
dbt_cloud_output = dbt_cloud.run_job_and_poll(
job_id=job_id,
cause=f"Materializing software-defined assets in Dagster run {context.run_id[:8]}",
cause=f"Materializing software-defined assets in Dagster run {context.run.run_id[:8]}",
steps_override=job_commands,
)

Expand Down
16 changes: 10 additions & 6 deletions python_modules/libraries/dagster-dbt/dagster_dbt/core/resources_v2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
from dagster import (
AssetCheckResult,
AssetCheckSeverity,
AssetExecutionContext,
AssetMaterialization,
AssetObservation,
AssetsDefinition,
Expand Down Expand Up @@ -827,7 +828,7 @@ def _get_unique_target_path(self, *, context: Optional[OpExecutionContext]) -> P
unique_id = str(uuid.uuid4())[:7]
path = unique_id
if context:
path = f"{context.op.name}-{context.run_id[:7]}-{unique_id}"
path = f"{context.op.name}-{context.run.run_id[:7]}-{unique_id}"

current_target_path = _get_dbt_target_path()

Expand Down Expand Up @@ -967,9 +968,12 @@ def my_dbt_op(dbt: DbtCliResource):
dbt_macro_args = {"key": "value"}
dbt.cli(["run-operation", "my-macro", json.dumps(dbt_macro_args)]).wait()
"""
op_context = (
context.op_execution_context if isinstance(context, AssetExecutionContext) else context
jamiedemaria marked this conversation as resolved.
Show resolved Hide resolved
)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would prefer that we just keep this named context.

Can we move this to be after we instantiate assets_def, but before the if context and assets_def is not None?

Suggested change
op_context = (
context.op_execution_context if isinstance(context, AssetExecutionContext) else context
)
context: Optional[OpExecutionContext] = (
context.op_execution_context if isinstance(context, AssetExecutionContext) else context
)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure for the rename, but _get_unique_target_name needs the op context too and is called before we get the assets_def. I just moved fetching assets_def up, but lmk if you'd prefer something else

dagster_dbt_translator = validate_opt_translator(dagster_dbt_translator)

target_path = target_path or self._get_unique_target_path(context=context)
target_path = target_path or self._get_unique_target_path(context=op_context)
env = {
**os.environ.copy(),
# Run dbt with unbuffered output.
Expand Down Expand Up @@ -1011,7 +1015,7 @@ def my_dbt_op(dbt: DbtCliResource):

selection_args: List[str] = []
dagster_dbt_translator = dagster_dbt_translator or DagsterDbtTranslator()
if context and assets_def is not None:
if op_context and assets_def is not None:
manifest, dagster_dbt_translator = get_manifest_and_translator_from_dbt_assets(
[assets_def]
)
Expand All @@ -1027,10 +1031,10 @@ def my_dbt_op(dbt: DbtCliResource):
env["DBT_INDIRECT_SELECTION"] = "empty"

selection_args = get_subset_selection_for_context(
context=context,
context=op_context,
manifest=manifest,
select=context.op.tags.get("dagster-dbt/select"),
exclude=context.op.tags.get("dagster-dbt/exclude"),
select=op_context.op.tags.get("dagster-dbt/select"),
exclude=op_context.op.tags.get("dagster-dbt/exclude"),
dagster_dbt_translator=dagster_dbt_translator,
)
else:
Expand Down