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 werkzeug.exceptions import HTTPException, NotFound, GatewayTimeout
9 from sqlalchemy.orm.attributes import InstrumentedAttribute
10 from coprs import app
11 from coprs.exceptions import (
12 AccessRestricted,
13 ActionInProgressException,
14 CoprHttpException,
15 InsufficientStorage,
16 ObjectNotFound,
17 )
18 from coprs.logic.complex_logic import ComplexLogic
19
20
21 apiv3_ns = flask.Blueprint("apiv3_ns", __name__, url_prefix="/api_3")
22
23
24
25 GET = ["GET"]
26 POST = ["POST"]
27 PUT = ["POST", "PUT"]
28 DELETE = ["POST", "DELETE"]
32 def query_params_decorator(f):
33 @wraps(f)
34 def query_params_wrapper(*args, **kwargs):
35 sig = inspect.signature(f)
36 params = [x for x in sig.parameters]
37 params = list(set(params) - {"args", "kwargs"})
38 for arg in params:
39 if arg not in flask.request.args:
40
41 if sig.parameters[arg].default == sig.parameters[arg].empty:
42 raise CoprHttpException("Missing argument {}".format(arg))
43 kwargs[arg] = flask.request.args.get(arg)
44 return f(*args, **kwargs)
45 return query_params_wrapper
46 return query_params_decorator
47
50 def pagination_decorator(f):
51 @wraps(f)
52 def pagination_wrapper(*args, **kwargs):
53 form = PaginationForm(flask.request.args)
54 if not form.validate():
55 raise CoprHttpException(form.errors)
56 kwargs.update(form.data)
57 return f(*args, **kwargs)
58 return pagination_wrapper
59 return pagination_decorator
60
63 def file_upload_decorator(f):
64 @wraps(f)
65 def file_upload_wrapper(*args, **kwargs):
66 if "json" in flask.request.files:
67 data = json.loads(flask.request.files["json"].read()) or {}
68 tuples = [(k, v) for k, v in data.items()]
69 flask.request.form = ImmutableMultiDict(tuples)
70 return f(*args, **kwargs)
71 return file_upload_wrapper
72 return file_upload_decorator
73
76 limit = wtforms.IntegerField("Limit", validators=[wtforms.validators.Optional()])
77 offset = wtforms.IntegerField("Offset", validators=[wtforms.validators.Optional()])
78 order = wtforms.StringField("Order by", validators=[wtforms.validators.Optional()])
79 order_type = wtforms.SelectField("Order type", validators=[wtforms.validators.Optional()],
80 choices=[("ASC", "ASC"), ("DESC", "DESC")], default="ASC")
81
82
83 -def get_copr(ownername=None, projectname=None):
88
91 LIMIT = None
92 OFFSET = 0
93 ORDER = "id"
94
95 - def __init__(self, query, model, limit=None, offset=None, order=None, order_type=None, **kwargs):
108
109
111 order_attr = getattr(self.model, self.order, None)
112 if not order_attr:
113 msg = "Cannot order by {}, {} doesn't have such property".format(
114 self.order, self.model.__tablename__)
115 raise CoprHttpException(msg)
116
117
118
119 if not isinstance(order_attr, InstrumentedAttribute):
120 raise CoprHttpException("Cannot order by {}".format(self.order))
121
122 order_fun = (lambda x: x)
123 if self.order_type == 'ASC':
124 order_fun = sqlalchemy.asc
125 elif self.order_type == 'DESC':
126 order_fun = sqlalchemy.desc
127
128 return (self.query.order_by(order_fun(order_attr))
129 .limit(self.limit)
130 .offset(self.offset))
131
132 @property
135
136 - def map(self, fun):
137 return [fun(x) for x in self.get()]
138
141
144 """
145 The normal `Paginator` class works with a SQLAlchemy query object and
146 therefore can do limits and ordering on database level, which is ideal.
147 However, in some special cases, we already have a list of objects fetched
148 from database and need to adjust it based on user pagination preferences,
149 hence this special case of `Paginator` class.
150
151 It isn't efficient, it isn't pretty. Please use `Paginator` if you can.
152 """
170
185 return wrapper
186