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, ActionInProgressException
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
39
41 return self.respond("The API request timeouted", 504)
42
45
47 response = flask.jsonify(error=message)
48 response.status_code = code
49 return response
50
57
60 def query_params_decorator(f):
61 @wraps(f)
62 def query_params_wrapper(*args, **kwargs):
63 sig = inspect.signature(f)
64 params = [x for x in sig.parameters]
65 params = list(set(params) - {"args", "kwargs"})
66 for arg in params:
67 if arg not in flask.request.args:
68
69 if sig.parameters[arg].default == sig.parameters[arg].empty:
70 raise CoprHttpException("Missing argument {}".format(arg))
71 kwargs[arg] = flask.request.args.get(arg)
72 return f(*args, **kwargs)
73 return query_params_wrapper
74 return query_params_decorator
75
78 def pagination_decorator(f):
79 @wraps(f)
80 def pagination_wrapper(*args, **kwargs):
81 form = PaginationForm(flask.request.args)
82 if not form.validate():
83 raise CoprHttpException(form.errors)
84 kwargs.update(form.data)
85 return f(*args, **kwargs)
86 return pagination_wrapper
87 return pagination_decorator
88
91 def file_upload_decorator(f):
92 @wraps(f)
93 def file_upload_wrapper(*args, **kwargs):
94 if "json" in flask.request.files:
95 data = json.loads(flask.request.files["json"].read()) or {}
96 tuples = [(k, v) for k, v in data.items()]
97 flask.request.form = ImmutableMultiDict(tuples)
98 return f(*args, **kwargs)
99 return file_upload_wrapper
100 return file_upload_decorator
101
104 limit = wtforms.IntegerField("Limit", validators=[wtforms.validators.Optional()])
105 offset = wtforms.IntegerField("Offset", validators=[wtforms.validators.Optional()])
106 order = wtforms.StringField("Order by", validators=[wtforms.validators.Optional()])
107 order_type = wtforms.SelectField("Order type", validators=[wtforms.validators.Optional()],
108 choices=[("ASC", "ASC"), ("DESC", "DESC")], default="ASC")
109
110
111 -def get_copr(ownername=None, projectname=None):
116
119 LIMIT = None
120 OFFSET = 0
121 ORDER = "id"
122
123 - def __init__(self, query, model, limit=None, offset=None, order=None, order_type=None, **kwargs):
136
137
139 if not hasattr(self.model, self.order):
140 raise CoprHttpException("Can order by: {}".format(self.order))
141
142 order_fun = (lambda x: x)
143 if self.order_type == 'ASC':
144 order_fun = sqlalchemy.asc
145 elif self.order_type == 'DESC':
146 order_fun = sqlalchemy.desc
147
148 return self.query.order_by(order_fun(self.order)).limit(self.limit).offset(self.offset)
149
150 @property
153
154 - def map(self, fun):
155 return [fun(x) for x in self.get()]
156
159
161 @wraps(f)
162 def wrapper(ownername, projectname, **kwargs):
163 copr = get_copr(ownername, projectname)
164 if not flask.g.user.can_edit(copr):
165 raise AccessRestricted(
166 "User '{0}' can not see permissions for project '{1}' "\
167 "(missing admin rights)".format(
168 flask.g.user.name,
169 '/'.join([ownername, projectname])
170 )
171 )
172 return f(copr, **kwargs)
173 return wrapper
174