Package coprs :: Package views :: Package apiv3_ns
[hide private]
[frames] | no frames]

Source Code for Package coprs.views.apiv3_ns

  1  import json 
  2  import flask 
  3  import wtforms 
  4  import sqlalchemy 
  5  import inspect 
  6  from functools import wraps 
  7  from werkzeug.datastructures import ImmutableMultiDict 
  8  from coprs import app 
  9  from coprs.exceptions import CoprHttpException 
 10  from coprs.logic.complex_logic import ComplexLogic 
 11   
 12   
 13  apiv3_ns = flask.Blueprint("apiv3_ns", __name__, url_prefix="/api_3") 
 14   
 15   
 16  # HTTP methods 
 17  GET = ["GET"] 
 18  POST = ["POST"] 
 19  PUT = ["POST", "PUT"] 
 20  DELETE = ["POST", "DELETE"] 
21 22 23 @app.errorhandler(CoprHttpException) 24 -def handle_copr_exception(error):
25 return handle_api_error(error.message, error.code)
26
27 28 @app.errorhandler(404) 29 -def page_not_found(error):
30 return handle_api_error("Such API endpoint doesn't exist", 404)
31
32 33 @app.errorhandler(500) 34 -def page_not_found(error):
35 return handle_api_error("Request wasn't successful, there is probably a bug in the API code.", 500)
36
37 38 -def handle_api_error(message, code):
39 response = flask.jsonify(error=message) 40 response.status_code = code 41 return response
42
43 44 -def query_params():
45 def query_params_decorator(f): 46 @wraps(f) 47 def query_params_wrapper(*args, **kwargs): 48 sig = inspect.signature(f) 49 params = [x for x in sig.parameters] 50 params = list(set(params) - {"args", "kwargs"}) 51 for arg in params: 52 if arg not in flask.request.args: 53 # If parameter has a default value, it is not required 54 if sig.parameters[arg].default == sig.parameters[arg].empty: 55 raise CoprHttpException("Missing argument {}".format(arg)) 56 kwargs[arg] = flask.request.args.get(arg) 57 return f(*args, **kwargs)
58 return query_params_wrapper 59 return query_params_decorator 60
61 62 -def pagination():
63 def pagination_decorator(f): 64 @wraps(f) 65 def pagination_wrapper(*args, **kwargs): 66 form = PaginationForm(flask.request.args) 67 if not form.validate(): 68 raise CoprHttpException(form.errors) 69 kwargs.update(form.data) 70 return f(*args, **kwargs)
71 return pagination_wrapper 72 return pagination_decorator 73
74 75 -def file_upload():
76 def file_upload_decorator(f): 77 @wraps(f) 78 def file_upload_wrapper(*args, **kwargs): 79 if "json" in flask.request.files: 80 data = json.loads(flask.request.files["json"].read()) or {} 81 tuples = [(k, v) for k, v in data.items()] 82 flask.request.form = ImmutableMultiDict(tuples) 83 return f(*args, **kwargs)
84 return file_upload_wrapper 85 return file_upload_decorator 86
87 88 -class PaginationForm(wtforms.Form):
89 limit = wtforms.IntegerField("Limit", validators=[wtforms.validators.Optional()]) 90 offset = wtforms.IntegerField("Offset", validators=[wtforms.validators.Optional()]) 91 order = wtforms.StringField("Order by", validators=[wtforms.validators.Optional()]) 92 order_type = wtforms.SelectField("Order type", validators=[wtforms.validators.Optional()], 93 choices=[("ASC", "ASC"), ("DESC", "DESC")], default="ASC")
94
95 96 -def get_copr(ownername=None, projectname=None):
97 request = flask.request 98 ownername = ownername or request.form.get("ownername") or request.json["ownername"] 99 projectname = projectname or request.form.get("projectname") or request.json["projectname"] 100 return ComplexLogic.get_copr_by_owner_safe(ownername, projectname)
101
102 103 -class Paginator(object):
104 LIMIT = None 105 OFFSET = 0 106 ORDER = "id" 107 ORDER_TYPE = "ASC" 108
109 - def __init__(self, query, model, limit=None, offset=None, order=None, order_type=None, **kwargs):
110 self.query = query 111 self.model = model 112 self.limit = limit or self.LIMIT 113 self.offset = offset or self.OFFSET 114 self.order = order or self.ORDER 115 self.order_type = order_type or self.ORDER_TYPE
116
117 - def get(self):
118 if not hasattr(self.model, self.order): 119 raise CoprHttpException("Can order by: {}".format(self.order)) 120 order_col = self.order 121 order_fun = sqlalchemy.desc if self.order_type == "DESC" else sqlalchemy.asc 122 return self.query.order_by(None).limit(None).offset(None)\ 123 .order_by(order_fun(order_col)).limit(self.limit).offset(self.offset)
124 125 @property
126 - def meta(self):
127 return {k: getattr(self, k) for k in ["limit", "offset", "order", "order_type"]}
128
129 - def map(self, fun):
130 return [fun(x) for x in self.get()]
131
132 - def to_dict(self):
133 return [x.to_dict() for x in self.get()]
134