import traceback from typing import List, Dict from flask_restful import Resource, abort, reqparse from app.glob import get_manager from app.utils import validate_url pub_parser = reqparse.RequestParser() pub_parser.add_argument('url', type=str) pub_parser.add_argument('file', type=str, default='') pub_parser.add_argument('force', type=bool, default=False) pub_parser_with_name = pub_parser.copy() pub_parser_with_name.add_argument('name', required=True, type=str, help='name required') url_req_parser = reqparse.RequestParser() url_req_parser.add_argument('url', type=str, required=True) remove_parser = reqparse.RequestParser() remove_parser.add_argument('force', type=bool, default=False) class Publications(Resource): def get(self) -> List[Dict[str, str]]: manager = get_manager() return manager.publications() def put(self): args = pub_parser_with_name.parse_args() validate_url(args.url) manager = get_manager() url = args.url if not args.force and args.name in manager.list(): abort(409, message=f'publication exists: {args.name}') try: url = manager.add(args.name, url, args.file, args.force) except Exception as e: abort(422, message=f'publication failed: {args.name}', traceback=traceback.format_exc()) return {'message': 'created', 'name': args.name, 'url': url}, 201 class Publication(Resource): def get(self, name: str): manager = get_manager() if name not in manager.list(): abort(404, message=f'publication not found: {name}') return { name: manager.get(name) } def put(self, name: str): args = pub_parser.parse_args() validate_url(args.url) manager = get_manager() url = args.url if not args.force and name in manager.list(): abort(409, message=f'publication exists: {name}') try: url = manager.add(name, url, args.file, args.force) except Exception as e: abort(422, message=f'publication failed: {name}', traceback=traceback.format_exc()) return {'message': 'created', 'name': name, 'url': url}, 201 def delete(self, name: str): args = remove_parser.parse_args() manager = get_manager() if name not in manager.list(): abort(404, message=f'publication not found: {name}') try: manager.remove(name, args.force) except Exception as e: abort(422, message='delete failed', traceback=traceback.format_exc()) return {'message': 'deleted', 'name': name} class PublicationURL(Resource): def get(self, name: str): manager = get_manager() if name not in manager.list(): abort(404, message=f'publication not found: {name}') return { 'url': manager.get(name)['url'] } def post(self, name: str): args = url_req_parser.parse_args() validate_url(args.url) manager = get_manager() url = args.url if name not in manager.list(): abort(404, message=f'publication not found: {name}') try: manager.set_url(name, url) except Exception as e: abort(422, message=f'set url failed', traceback=traceback.format_exc()) return {'message': 'success', 'name': name, 'url': url}