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, ObjectNotFound, AccessRestricted
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
17 GET = ["GET"]
18 POST = ["POST"]
19 PUT = ["POST", "PUT"]
20 DELETE = ["POST", "DELETE"]
25 if isinstance(error, ObjectNotFound):
26 return self.respond(str(error), 404)
27 return self.respond("Such API endpoint doesn't exist", 404)
28
31
34
36 return self.respond("Request wasn't successful, there is probably a bug in the API code.", 500)
37
39 return self.respond("The API request timeouted", 504)
40
43
45 response = flask.jsonify(error=message)
46 response.status_code = code
47 return response
48
55
58 def query_params_decorator(f):
59 @wraps(f)
60 def query_params_wrapper(*args, **kwargs):
61 sig = inspect.signature(f)
62 params = [x for x in sig.parameters]
63 params = list(set(params) - {"args", "kwargs"})
64 for arg in params:
65 if arg not in flask.request.args:
66
67 if sig.parameters[arg].default == sig.parameters[arg].empty:
68 raise CoprHttpException("Missing argument {}".format(arg))
69 kwargs[arg] = flask.request.args.get(arg)
70 return f(*args, **kwargs)
71 return query_params_wrapper
72 return query_params_decorator
73
76 def pagination_decorator(f):
77 @wraps(f)
78 def pagination_wrapper(*args, **kwargs):
79 form = PaginationForm(flask.request.args)
80 if not form.validate():
81 raise CoprHttpException(form.errors)
82 kwargs.update(form.data)
83 return f(*args, **kwargs)
84 return pagination_wrapper
85 return pagination_decorator
86
89 def file_upload_decorator(f):
90 @wraps(f)
91 def file_upload_wrapper(*args, **kwargs):
92 if "json" in flask.request.files:
93 data = json.loads(flask.request.files["json"].read()) or {}
94 tuples = [(k, v) for k, v in data.items()]
95 flask.request.form = ImmutableMultiDict(tuples)
96 return f(*args, **kwargs)
97 return file_upload_wrapper
98 return file_upload_decorator
99
102 limit = wtforms.IntegerField("Limit", validators=[wtforms.validators.Optional()])
103 offset = wtforms.IntegerField("Offset", validators=[wtforms.validators.Optional()])
104 order = wtforms.StringField("Order by", validators=[wtforms.validators.Optional()])
105 order_type = wtforms.SelectField("Order type", validators=[wtforms.validators.Optional()],
106 choices=[("ASC", "ASC"), ("DESC", "DESC")], default="ASC")
107
108
109 -def get_copr(ownername=None, projectname=None):
114
117 LIMIT = None
118 OFFSET = 0
119 ORDER = "id"
120 ORDER_TYPE = "ASC"
121
122 - def __init__(self, query, model, limit=None, offset=None, order=None, order_type=None, **kwargs):
129
137
138 @property
141
142 - def map(self, fun):
143 return [fun(x) for x in self.get()]
144
147
149 @wraps(f)
150 def wrapper(ownername, projectname, **kwargs):
151 copr = get_copr(ownername, projectname)
152 if not flask.g.user.can_edit(copr):
153 raise AccessRestricted(
154 "User '{0}' can not see permissions for project '{1}' "\
155 "(missing admin rights)".format(
156 flask.g.user.name,
157 '/'.join([ownername, projectname])
158 )
159 )
160 return f(copr, **kwargs)
161 return wrapper
162