How to trigger failure if arglist doesn't exist as a group step

[chrisfunston]

I'm just not sure how to do this properly, or if there's an easy way to do so. I'm taking in a list of build arguments, and if someone were to put in a step that doesn't exist, pypyr doesn't call it, but also doesn't trigger a 'failure'.

For example:

context_parser: pypyr.parser.list
steps:
  - name: pypyr.steps.call
    foreach: '{argList}'
    in:
      call:
        groups: '{i}'
        success: success_group
        failure: failure_group

testing:
  - name: pypyr.steps.echo
    in:
      echoMe: "In testing"

success_group:
  - name: pypyr.steps.echo
    in:
      echoMe: "All working!"

failure_group:
  - name: pypyr.steps.echo
    in:
      echoMe: "I'm right here! Something went wrong!"

If I run pypyr build asdf, I get the following:

PS C:\pypyr> pypyr test asdf
All working!

Is there a way to force this failure, or alternatively, is there a way I can validate a group step to a list that pypyr has loaded?

Thanks!

[yaythomas]

Hi @chrisfunston,

This is by design. . . Or at least I remember I had some sort of good reason for not throwing a hard error and stopping the pipeline if a called group name doesn't exist, but to my embarrasment, now I can't remember why! I'll have a bit more of a think to see if I can ferret out the use-case I had in mind.

In the meantime, you can get the effect you want like this:

context_parser: pypyr.parser.list
steps:
  - name: pypyr.steps.py
    in:
      pycode: |
        #  this is the top-level dict for your pipeline
        pipeline = context.current_pipeline.pipeline_definition.pipeline
        
        for group in context['argList']:
          if group not in pipeline:
            raise ValueError(f'group {group} does not exist in pipeline!')

  - name: pypyr.steps.call
    foreach: '{argList}'
    in:
      call:
        groups: '{i}'
        success: success_group
        failure: failure_group

testing:
  - name: pypyr.steps.echo
    in:
      echoMe: "In testing"

success_group:
  - name: pypyr.steps.echo
    in:
      echoMe: "All working!"

failure_group:
  - name: pypyr.steps.echo
    in:
      echoMe: This is the error handler that will run if the `call` step or anything it calls fails.

on_failure:
  - name: pypyr.steps.echo
    in:
      echoMe: This is the general catch-all err handler that will run if the requested group doesn't exist in the pipeline.

For a bit of context, so you understand what's going here - we're grabbing the current pipeline object from the context object in the py step with the pycode style input that gives us access to the actual content object - and then going to the pipeline's definition. This definition is basically the yaml that pypyr loaded for the pipeline - and it's represented in something that behaves just like a dict. At the top level of this dict the keys are the step-group names.

If you were keen you could make a custom step that pretty much contains the code in the py-step, and then you could simplify your pipeline.

If you create ./mydir/validate_groups.py:

# ./mydir/validate_groups.p
import logging

from pypyr.context import Context
from pypyr.errors import Error

logger = logging.getLogger(__name__)


def run_step(context: Context) -> None:
    logger.debug("started")

    #  this is the top-level dict for your pipeline
    pipeline = context.current_pipeline.pipeline_definition.pipeline

    for group in context['argList']:
        logger.debug(f"checking if {group} exists in pipeline's step groups")
        if group not in pipeline:
            raise Error(f'group {group} does not exist in pipeline!')

    logger.debug("done")

And then the pipeline can just use the re-usable step like this:

context_parser: pypyr.parser.list
steps:
  - mydir.validate_groups
  - name: pypyr.steps.call
    foreach: '{argList}'
    in:
      call:
        groups: '{i}'
        success: success_group
        failure: failure_group

[chrisfunston]

Thank you so much for the response! Works like a charm, it's exactly what I was looking for. Cheers!

[yaythomas]

The way I normally do it myself, is to have validation of input arguments separate at the beginning with assert:

context_parser: pypyr.parser.list
steps:
  - name: pypyr.steps.assert
    comment: do this if there HAS to be >0 input args
    in:
      assert:
        this: '{argList}'
        msg: You must specify an input argument

  - name: pypyr.steps.assert
    comment: verify input arguments
    foreach: '{argList}'
    in:
      valid_inputs:
        - group1
        - group2
      assert:
        this: !py i in valid_inputs
        msg: invalid input argument - {i}


  - name: pypyr.steps.call
    run: '{argList}'
    in:
      call: '{argList}'

Little note: you don't actually need the foreach on your call step - the abbreviated call input form and the full call.groups input both support lists directly:

  - name: pypyr.steps.call
    run: '{argList}'
    in:
      call: '{argList}'

  - name: pypyr.steps.call
    run: '{argList}'
    in:
      call:
       groups: '{argList}'
       success: success_group
       failure: failure_group