Package coprs :: Package rest_api :: Module util
[hide private]
[frames] | no frames]

Source Code for Module coprs.rest_api.util

  1  # coding: utf-8 
  2  import json 
  3   
  4  from flask import Response, url_for, Blueprint 
  5  import sqlalchemy.orm.exc 
  6   
  7  from flask_restful.reqparse import Argument, RequestParser 
  8   
  9  from .exceptions import ObjectNotFoundError, MalformedRequest 
 10  from .schemas import AllowedMethodSchema 
 11   
 12   
13 -class AllowedMethod(object):
14 - def __init__(self, method, doc, require_auth=True, params=None):
15 self.method = method 16 self.doc = doc 17 self.require_auth = require_auth 18 self.params = params or []
19 20
21 -def render_allowed_method(method, doc, require_auth=True, params=None):
22 return AllowedMethodSchema().dump(AllowedMethod(method, doc, require_auth, params))[0]
23 24
25 -def get_one_safe(query, msg=None, data=None):
26 """ 27 :type query: sqlalchemy.Query 28 :param str msg: message used in error when object not found 29 :param Any data: any serializable data to include into error 30 :raises ObjectNotFoundError: when query failed to return anything 31 """ 32 try: 33 return query.one() 34 except sqlalchemy.orm.exc.NoResultFound: 35 raise ObjectNotFoundError(msg=msg, data=data)
36 37
38 -def json_loads_safe(raw, data_on_error=None):
39 try: 40 return json.loads(raw) 41 except ValueError: 42 raise MalformedRequest(data_on_error or 43 "Failed to deserialize json string")
44 45
46 -def mm_deserialize(schema, json_string):
47 try: 48 result = schema.loads(json_string) 49 except ValueError as err: 50 raise MalformedRequest( 51 msg="Failed to parse request: {}".format(err), 52 data={"request_string": json_string} 53 ) 54 55 if result.errors: 56 raise MalformedRequest( 57 msg="Failed to parse request: Validation Error", 58 data={ 59 "validation_errors": result.errors 60 } 61 ) 62 63 return result
64 65
66 -def mm_serialize_one(schema, obj):
67 return schema().dump(obj)[0]
68 69
70 -class MyArg(Argument):
71 - def handle_validation_error(self, error, bundle_errors):
72 # dirty monkey patching, better to switch to webargs 73 # bundle errors are ignored 74 data = {u"error": unicode(error)} 75 if self.help: 76 data["help"] = self.help 77 raise MalformedRequest( 78 "Failed to validate query parameter: {}".format(self.name), 79 data=data 80 )
81 82
83 -def get_request_parser():
84 return RequestParser(argument_class=MyArg)
85 86
87 -def arg_bool(value):
88 """ 89 :param str value: value to convert 90 :rtype: bool 91 :raises ValueError: 92 """ 93 true_values = ["true", "yes", "1"] 94 false_values = ["false", "no", "0"] 95 value = str(value).strip().lower() 96 97 if value in true_values: 98 return True 99 elif value in false_values: 100 return False 101 102 raise ValueError("Value `{}` doesn't belong to either true of false sets. " 103 "Allowed values: {} and {}" 104 .format(value, true_values, false_values))
105