You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
It seems to be nice to be able to log out specific expectations that failed, when the Operator fails. I am not sure this should be internalized, so instead, perhaps it makes sense to set a callback function on fail?
I have something as naive as this in mind:
def generate_list_failed_expectations(r) -> str:
"""Note - only for validation results
Input:
r (reat_expectations.validation_operators.types.validation_operator_result.ValidationOperatorResult)
Returns: str
"""
from itertools import groupby
lines = ["Validation with Great Expectations failed."]
for batch, result in r["run_results"].items():
lines.append(f"{batch}")
failed = [
r.expectation_config
for r in result["validation_result"].results
if not r.success
]
for col, group_failed in groupby(failed, key=lambda k: k["kwargs"]["column"]):
lines.append(f" {col}")
for test in group_failed:
lines.append(f' - {test["expectation_type"]}')
raise AirflowException("\n".join(lines))
The text was updated successfully, but these errors were encountered:
It seems to be nice to be able to log out specific expectations that failed, when the Operator fails. I am not sure this should be internalized, so instead, perhaps it makes sense to set a callback function on fail?
I have something as naive as this in mind:
The text was updated successfully, but these errors were encountered: