-
Notifications
You must be signed in to change notification settings - Fork 1.7k
[asset health] include freshness in asset health #29268
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.
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
Changes from 5 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -9,6 +9,7 @@ | |
from collections.abc import Iterator, Mapping, Sequence | ||
from contextlib import contextmanager | ||
from copy import deepcopy | ||
from datetime import timedelta | ||
from typing import Optional, TypeVar, Union | ||
|
||
from dagster import ( | ||
|
@@ -96,6 +97,7 @@ | |
from dagster._core.definitions.events import Failure | ||
from dagster._core.definitions.executor_definition import in_process_executor | ||
from dagster._core.definitions.external_asset import external_asset_from_spec | ||
from dagster._core.definitions.freshness import InternalFreshnessPolicy | ||
from dagster._core.definitions.freshness_policy import FreshnessPolicy | ||
from dagster._core.definitions.job_definition import JobDefinition | ||
from dagster._core.definitions.metadata import MetadataValue | ||
|
@@ -1677,7 +1679,12 @@ def the_op(): | |
the_op() | ||
|
||
|
||
@asset(owners=["[email protected]", "team:team1"]) | ||
@asset( | ||
owners=["[email protected]", "team:team1"], | ||
internal_freshness_policy=InternalFreshnessPolicy.time_window( | ||
fail_window=timedelta(minutes=10), warn_window=timedelta(minutes=5) | ||
), | ||
) | ||
def asset_1(): | ||
yield Output(3) | ||
|
||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -37,7 +37,7 @@ | |
) | ||
from dagster._core.definitions.data_version import extract_data_provenance_from_entry | ||
from dagster._core.definitions.events import AssetKey, AssetObservation | ||
from dagster._core.definitions.freshness import FreshnessStateEvaluation | ||
from dagster._core.definitions.freshness import FreshnessStateEvaluation, FreshnessStateRecord | ||
from dagster._core.definitions.partition_key_range import PartitionKeyRange | ||
from dagster._core.errors import ( | ||
DagsterHomeNotSetError, | ||
|
@@ -3395,6 +3395,10 @@ def report_runless_asset_event( | |
), | ||
) | ||
|
||
def get_entity_freshness_state(self, entity_key: AssetKey) -> Optional[FreshnessStateRecord]: | ||
warnings.warn("`get_entity_freshness_state` is not yet implemented for OSS.") | ||
return None | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. does this need any warning or error message to signify that it's not implemented in OSS / is cloud only? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. since the method isn't marked public i think it's probably fine? in other similar cases we will usually return |
||
|
||
def get_asset_check_support(self) -> "AssetCheckInstanceSupport": | ||
from dagster._core.storage.asset_check_execution_record import AssetCheckInstanceSupport | ||
|
||
|
Uh oh!
There was an error while loading. Please reload this page.