https://github.com/stitchfix/hamilton logo
Join Slack
Powered by
Hello, I want to validate the output of a functio...
# hamilton-help
p
Hello, I want to validate the output of a function, that I am later using in pipe_output as a 
step
. I wanted to use 
@check_output_custom
, but the validation is not propagated. Is there a way to do this? Example:
Copy code
@check_output_custom(CustomValidator())
def foo(...) -> ...:
    ...
    return result

@pipe_output(
    step(foo).named('foo_applied'),
    ...,
)
def bar(...) -> ...:
    return result
Validation is not applied to 
foo_applied
Thank you in advance for reply ---------------------------------------------------------- Also is there a way of acting upon failure? E.g I know, that if validation fails in one of the previous steps, than I want to return a flag 
inference failed
at the end of the flow, on top of exiting it. Basically some functionality for early termination of a flow
s
@check_output_*
only works on first class functions in the hamilton DAG, not "steps". Want to try the following?
Copy code
def foo(...) -> ...:
    ...
    return result

@check_output_custom(CustomValidator(), target_="foo_applied")
@pipe_output(
    step(foo).named('foo_applied'),
    ...,
)
def transition_cap(...) -> ...:
    return result
> Also is there a way of acting upon failure? > > E.g I know, that if validation fails in one of the previous steps, than I want to return a flag 
inference failed
at the end of the flow, on top of exiting it. What is the current behavior you're observing? IIRC it should fail hard, i.e. throw an exception if things fail and stop all execution -- is that what you want? CC @Elijah Ben Izzy
e
Re: error management -- you can rely on it raising this exception if it fails: https://github.com/DAGWorks-Inc/hamilton/blob/main/hamilton/data_quality/base.py#L10. Then you can catch that, which will have some error information, and you can do what you want. That said, it's not programmatic (it's formatted as a string), but you could also build a hook that fails when calling it on warning, using the output -- can show you what that might look liek if it's interesting. But yeah to @Stefan Krawczykā€™s point -- what are you observing and how does that differ from your expectations?
p
Thank you for swift reply. > Want to try the following? [...] That would not be suitable, as the step(foo) is applied across many different nodes, and I want the validation to always follow this step. > What is the current behavior you're observing? IIRC it should fail hard, i.e. throw an exception if things fail and stop all execution -- is that what you want? Ideally: If I have A -> B -> C -> D And B fails validation, I would like C and D (some downstream nodes) to return a default node value, that would be possible to set. This is because, once, I know that B does not produce satisfactory results, The rest of the flow does not need to be executed.
Then you can catch that, which will have some error information, and you can do what you want.
I did try this, however logging the error does not seem appealing, as I would like this to be part of expected functionality. Maybe it is wrong design choice on my side, but I believe it can have many use cases. To give an example: Imagine a data flow that takes as input df of cars, uses nodes to filter the cars, and calculates some metrics (e.g total price, or total mass, or some other attribute), of all of the filtered cars. If there are many filters, it would be nice if all of the filters would assert that output contains at least one car, otherwise, the execution is stopped prematurely at that filter node, and all of the 
calculate_metric
nodes should return a default value of 0 (or whatever was specified for them).
I specifically need to do something along the lines of:
Copy code
@check_output_custom(CarsExistValidator())
def filter_color(...)

def filter_horse_power(...)

def filter_type(...)

def total_price(...)

def total_mass(...)

@pipe_output(
  step(filter_color),
  step(filter_type),
  step(filter_hp),
  step(total_price),
def total_mass_red_suv_200hp(cars: df)

@pipe_output(
  step(filter_color),
  step(total_mass)
  step(filter_type),
  step(filter_hp),
  step(total_price)
def total_price_blue_light_sport_cheap(cars: df)
And so If there are no blue cars, instead of propagating redundant execution, I would like the total_mass to return 0 in case 
CarsExistValidator()
fails.
s
That would not be suitable, as the step(foo) is applied across many different nodes, and I want the validation to always follow this step.
@Piotr Bieszczad makes sense. Mind opening up an issue here? Would have to dig into the code to determine the feasibility of this. But at a high level it makes sense. To get you immediately unblocked I think there's a few paths. Before enumerating them, are you using any "adapters" with your code right now?
p
Okay, I will open an issue. I didn't really look into adapters yet. I have only used UI adapter so far. 
GracefulErrorAdapter
sounds promising. I will read up on that.
šŸ‘ 1
s
Cool. So I think the most straightforward thing to unblock you right now would be to write your own python decorator around the 
foo
function. It can take in the validator and exercise it for each invocation of that function. It can then throw a specific error that the graceful error adapter could catch / handle.
Are you familiar with how python decorators work?
Open in Slack
PreviousNext
Powered by