recipe as script simple sketch

__init__.py

import argparse
import functools
import inspect
import json
import sys


def recipe_inputs(schema, kw='inputs'):
    def decorator(recipe_func):
        @functools.wraps(recipe_func)
        def wrapper(workflow, inputs):
            validated_inputs = schema(inputs)
            return recipe_func(workflow, validated_inputs)
        wrapper.schema = schema
        return wrapper
    return decorator


def recipe_cli(recipe):
    def decorator(cli_func):
        arg_parser = argparse.ArgumentParser(
            description=inspect.getdoc(cli_func))
        arg_parser.add_argument(
            '--in_json', default=sys.stdin, help=str(recipe.schema.schema))
        
        # cosmos setup

        @functools.wraps(cli_func)
        def wrapper():
            args = arg_parser.parse_args()
            inputs = json.load(args.in_json)
            
            # cosmos run
            # we could optionally pass workflow outputs into the cli function for custom post assertions or exits
            recipe('workflow', inputs)
           
        return wrapper

    return decorator

__main__.py

from . import recipe_cli, recipes


@recipe_cli(recipes.bee_tuna)
def bee_tuna_cli():
    pass


if __name__ == "__main__":
    bee_tuna_cli()

exec_recipe.md

• keeps the concerns more separate
• allows for multiple recipes to a single module
• doesn't need to be executed by running the module directly and adding the ifmain pattern in too many places
• python -m module executability (even without being an installable package)
• plays better with pipe as py package
• setuptools can provide entrypoint definitions - a neat feature of py packages
• composability of cli commands into groups
• can play well with argparse or click

recipes.py

from voluptuous import Schema, Required, All, Length, Range

from . import recipe_inputs


@recipe_inputs(schema=Schema({
    Required('q'): All(basestring, Length(min=1)),
    Required('per_page', default=5): All(int, Range(min=1, max=20)),
    'page': All(int, Range(min=0)),
}))
def bee_tuna(workflow, inputs):
    print(workflow)
    print(inputs)