Skip to content

Instantly share code, notes, and snippets.

@p7k
Last active Feb 12, 2017
Embed
What would you like to do?
recipe as script simple sketch
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
from . import recipe_cli, recipes
@recipe_cli(recipes.bee_tuna)
def bee_tuna_cli():
pass
if __name__ == "__main__":
bee_tuna_cli()
  • 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
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment