tricu

A lazy functional language based on Tree Calculus
Log | Files | Refs | README | LICENSE

cli.py (2169B)

      1 import argparse
      2 import json
      3 import sys
      4 
      5 from . import Session, WorkspaceExecutor, ProtocolError, identifier
      6 
      7 
      8 def grant(text):
      9     identity, separator, stage = text.partition(':')
     10     if not separator or len(identity) != 64 or any(c not in '0123456789abcdef' for c in identity):
     11         raise argparse.ArgumentTypeError('approval must be DEFINITION_SHA256:STAGE')
     12     try:
     13         identifier(stage)
     14     except ProtocolError as error:
     15         raise argparse.ArgumentTypeError(str(error)) from error
     16     return identity, stage
     17 
     18 
     19 def main():
     20     parser = argparse.ArgumentParser(description='Inspectable pipelines with explicitly authorized local effects')
     21     parser.add_argument('command', choices=['explain', 'execute'])
     22     parser.add_argument('--message', default='hello pipeline\n')
     23     parser.add_argument('--approval', action='store_true', help='insert approval before deploy')
     24     parser.add_argument('--workspace')
     25     parser.add_argument('--allow', action='append', choices=['write', 'check', 'publish'], default=[])
     26     parser.add_argument('--approve', type=grant, action='append', default=[])
     27     args = parser.parse_args()
     28     if args.command == 'execute' and not args.workspace:
     29         parser.error('execute requires an existing --workspace')
     30     if args.command == 'explain' and (args.workspace or args.allow or args.approve):
     31         parser.error('explain accepts no execution authority')
     32     try:
     33         with Session() as session, session.context.scope() as roots:
     34             plan = session.plan(roots, args.message, args.approval)
     35             if args.command == 'explain':
     36                 result = session.describe(roots, plan)
     37             else:
     38                 with WorkspaceExecutor(args.workspace, args.allow, args.approve) as executor:
     39                     result = session.execute(roots, plan, executor)
     40         print(json.dumps(result, ensure_ascii=False))
     41         return 0 if result.get('outcome', 'completed') == 'completed' else 1
     42     except (OSError, ValueError, RuntimeError) as error:
     43         print(json.dumps(dict(outcome='error', reason=str(error)), ensure_ascii=True))
     44         return 2
     45 
     46 
     47 if __name__ == '__main__':
     48     sys.exit(main())