9688f60c3411a176c27cfe881db595655f3e3606
[osm/NBI.git] / osm_nbi / base_topic.py
1 # -*- coding: utf-8 -*-
2
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12 # implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15
16 import logging
17 from uuid import uuid4
18 from http import HTTPStatus
19 from time import time
20 from osm_common.dbbase import deep_update_rfc7396
21 from osm_nbi.validation import validate_input, ValidationError, is_valid_uuid
22
23 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
24
25
26 class EngineException(Exception):
27
28 def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
29 self.http_code = http_code
30 super(Exception, self).__init__(message)
31
32
33 def get_iterable(input_var):
34 """
35 Returns an iterable, in case input_var is None it just returns an empty tuple
36 :param input_var: can be a list, tuple or None
37 :return: input_var or () if it is None
38 """
39 if input_var is None:
40 return ()
41 return input_var
42
43
44 def versiontuple(v):
45 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
46 filled = []
47 for point in v.split("."):
48 filled.append(point.zfill(8))
49 return tuple(filled)
50
51
52 class BaseTopic:
53 # static variables for all instance classes
54 topic = None # to_override
55 topic_msg = None # to_override
56 schema_new = None # to_override
57 schema_edit = None # to_override
58 multiproject = True # True if this Topic can be shared by several projects. Then it contains _admin.projects_read
59
60 # Alternative ID Fields for some Topics
61 alt_id_field = {
62 "projects": "name",
63 "users": "username",
64 "roles": "name"
65 }
66
67 def __init__(self, db, fs, msg):
68 self.db = db
69 self.fs = fs
70 self.msg = msg
71 self.logger = logging.getLogger("nbi.engine")
72
73 @staticmethod
74 def id_field(topic, value):
75 """Returns ID Field for given topic and field value"""
76 if topic in BaseTopic.alt_id_field.keys() and not is_valid_uuid(value):
77 return BaseTopic.alt_id_field[topic]
78 else:
79 return "_id"
80
81 @staticmethod
82 def _remove_envelop(indata=None):
83 if not indata:
84 return {}
85 return indata
86
87 def _validate_input_new(self, input, force=False):
88 """
89 Validates input user content for a new entry. It uses jsonschema. Some overrides will use pyangbind
90 :param input: user input content for the new topic
91 :param force: may be used for being more tolerant
92 :return: The same input content, or a changed version of it.
93 """
94 if self.schema_new:
95 validate_input(input, self.schema_new)
96 return input
97
98 def _validate_input_edit(self, input, force=False):
99 """
100 Validates input user content for an edition. It uses jsonschema. Some overrides will use pyangbind
101 :param input: user input content for the new topic
102 :param force: may be used for being more tolerant
103 :return: The same input content, or a changed version of it.
104 """
105 if self.schema_edit:
106 validate_input(input, self.schema_edit)
107 return input
108
109 @staticmethod
110 def _get_project_filter(session):
111 """
112 Generates a filter dictionary for querying database, so that only allowed items for this project can be
113 addressed. Only propietary or public can be used. Allowed projects are at _admin.project_read/write. If it is
114 not present or contains ANY mean public.
115 :param session: contains:
116 project_id: project list this session has rights to access. Can be empty, one or several
117 set_project: items created will contain this project list
118 force: True or False
119 public: True, False or None
120 method: "list", "show", "write", "delete"
121 admin: True or False
122 :return: dictionary with project filter
123 """
124 p_filter = {}
125 project_filter_n = []
126 project_filter = list(session["project_id"])
127
128 if session["method"] not in ("list", "delete"):
129 if project_filter:
130 project_filter.append("ANY")
131 elif session["public"] is not None:
132 if session["public"]:
133 project_filter.append("ANY")
134 else:
135 project_filter_n.append("ANY")
136
137 if session.get("PROJECT.ne"):
138 project_filter_n.append(session["PROJECT.ne"])
139
140 if project_filter:
141 if session["method"] in ("list", "show", "delete") or session.get("set_project"):
142 p_filter["_admin.projects_read.cont"] = project_filter
143 else:
144 p_filter["_admin.projects_write.cont"] = project_filter
145 if project_filter_n:
146 if session["method"] in ("list", "show", "delete") or session.get("set_project"):
147 p_filter["_admin.projects_read.ncont"] = project_filter_n
148 else:
149 p_filter["_admin.projects_write.ncont"] = project_filter_n
150
151 return p_filter
152
153 def check_conflict_on_new(self, session, indata):
154 """
155 Check that the data to be inserted is valid
156 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
157 :param indata: data to be inserted
158 :return: None or raises EngineException
159 """
160 pass
161
162 def check_conflict_on_edit(self, session, final_content, edit_content, _id):
163 """
164 Check that the data to be edited/uploaded is valid
165 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
166 :param final_content: data once modified. This method may change it.
167 :param edit_content: incremental data that contains the modifications to apply
168 :param _id: internal _id
169 :return: None or raises EngineException
170 """
171 if not self.multiproject:
172 return
173 # Change public status
174 if session["public"] is not None:
175 if session["public"] and "ANY" not in final_content["_admin"]["projects_read"]:
176 final_content["_admin"]["projects_read"].append("ANY")
177 final_content["_admin"]["projects_write"].clear()
178 if not session["public"] and "ANY" in final_content["_admin"]["projects_read"]:
179 final_content["_admin"]["projects_read"].remove("ANY")
180
181 # Change project status
182 if session.get("set_project"):
183 for p in session["set_project"]:
184 if p not in final_content["_admin"]["projects_read"]:
185 final_content["_admin"]["projects_read"].append(p)
186
187 def check_unique_name(self, session, name, _id=None):
188 """
189 Check that the name is unique for this project
190 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
191 :param name: name to be checked
192 :param _id: If not None, ignore this entry that are going to change
193 :return: None or raises EngineException
194 """
195 if not self.multiproject:
196 _filter = {}
197 else:
198 _filter = self._get_project_filter(session)
199 _filter["name"] = name
200 if _id:
201 _filter["_id.neq"] = _id
202 if self.db.get_one(self.topic, _filter, fail_on_empty=False, fail_on_more=False):
203 raise EngineException("name '{}' already exists for {}".format(name, self.topic), HTTPStatus.CONFLICT)
204
205 @staticmethod
206 def format_on_new(content, project_id=None, make_public=False):
207 """
208 Modifies content descriptor to include _admin
209 :param content: descriptor to be modified
210 :param project_id: if included, it add project read/write permissions. Can be None or a list
211 :param make_public: if included it is generated as public for reading.
212 :return: op_id: operation id on asynchronous operation, None otherwise. In addition content is modified
213 """
214 now = time()
215 if "_admin" not in content:
216 content["_admin"] = {}
217 if not content["_admin"].get("created"):
218 content["_admin"]["created"] = now
219 content["_admin"]["modified"] = now
220 if not content.get("_id"):
221 content["_id"] = str(uuid4())
222 if project_id is not None:
223 if not content["_admin"].get("projects_read"):
224 content["_admin"]["projects_read"] = list(project_id)
225 if make_public:
226 content["_admin"]["projects_read"].append("ANY")
227 if not content["_admin"].get("projects_write"):
228 content["_admin"]["projects_write"] = list(project_id)
229 return None
230
231 @staticmethod
232 def format_on_edit(final_content, edit_content):
233 """
234 Modifies final_content to admin information upon edition
235 :param final_content: final content to be stored at database
236 :param edit_content: user requested update content
237 :return: operation id, if this edit implies an asynchronous operation; None otherwise
238 """
239 if final_content.get("_admin"):
240 now = time()
241 final_content["_admin"]["modified"] = now
242 return None
243
244 def _send_msg(self, action, content):
245 if self.topic_msg:
246 content.pop("_admin", None)
247 self.msg.write(self.topic_msg, action, content)
248
249 def check_conflict_on_del(self, session, _id, db_content):
250 """
251 Check if deletion can be done because of dependencies if it is not force. To override
252 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
253 :param _id: internal _id
254 :param db_content: The database content of this item _id
255 :return: None if ok or raises EngineException with the conflict
256 """
257 pass
258
259 @staticmethod
260 def _update_input_with_kwargs(desc, kwargs):
261 """
262 Update descriptor with the kwargs. It contains dot separated keys
263 :param desc: dictionary to be updated
264 :param kwargs: plain dictionary to be used for updating.
265 :return: None, 'desc' is modified. It raises EngineException.
266 """
267 if not kwargs:
268 return
269 try:
270 for k, v in kwargs.items():
271 update_content = desc
272 kitem_old = None
273 klist = k.split(".")
274 for kitem in klist:
275 if kitem_old is not None:
276 update_content = update_content[kitem_old]
277 if isinstance(update_content, dict):
278 kitem_old = kitem
279 elif isinstance(update_content, list):
280 kitem_old = int(kitem)
281 else:
282 raise EngineException(
283 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem))
284 update_content[kitem_old] = v
285 except KeyError:
286 raise EngineException(
287 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old))
288 except ValueError:
289 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
290 k, kitem))
291 except IndexError:
292 raise EngineException(
293 "Invalid query string '{}'. Index '{}' out of range".format(k, kitem_old))
294
295 def show(self, session, _id):
296 """
297 Get complete information on an topic
298 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
299 :param _id: server internal id
300 :return: dictionary, raise exception if not found.
301 """
302 if not self.multiproject:
303 filter_db = {}
304 else:
305 filter_db = self._get_project_filter(session)
306 # To allow project&user addressing by name AS WELL AS _id
307 filter_db[BaseTopic.id_field(self.topic, _id)] = _id
308 return self.db.get_one(self.topic, filter_db)
309 # TODO transform data for SOL005 URL requests
310 # TODO remove _admin if not admin
311
312 def get_file(self, session, _id, path=None, accept_header=None):
313 """
314 Only implemented for descriptor topics. Return the file content of a descriptor
315 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
316 :param _id: Identity of the item to get content
317 :param path: artifact path or "$DESCRIPTOR" or None
318 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
319 :return: opened file or raises an exception
320 """
321 raise EngineException("Method get_file not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR)
322
323 def list(self, session, filter_q=None):
324 """
325 Get a list of the topic that matches a filter
326 :param session: contains the used login username and working project
327 :param filter_q: filter of data to be applied
328 :return: The list, it can be empty if no one match the filter.
329 """
330 if not filter_q:
331 filter_q = {}
332 if self.multiproject:
333 filter_q.update(self._get_project_filter(session))
334
335 # TODO transform data for SOL005 URL requests. Transform filtering
336 # TODO implement "field-type" query string SOL005
337 return self.db.get_list(self.topic, filter_q)
338
339 def new(self, rollback, session, indata=None, kwargs=None, headers=None):
340 """
341 Creates a new entry into database.
342 :param rollback: list to append created items at database in case a rollback may to be done
343 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
344 :param indata: data to be inserted
345 :param kwargs: used to override the indata descriptor
346 :param headers: http request headers
347 :return: _id, op_id:
348 _id: identity of the inserted data.
349 op_id: operation id if this is asynchronous, None otherwise
350 """
351 try:
352 content = self._remove_envelop(indata)
353
354 # Override descriptor with query string kwargs
355 self._update_input_with_kwargs(content, kwargs)
356 content = self._validate_input_new(content, force=session["force"])
357 self.check_conflict_on_new(session, content)
358 op_id = self.format_on_new(content, project_id=session["project_id"], make_public=session["public"])
359 _id = self.db.create(self.topic, content)
360 rollback.append({"topic": self.topic, "_id": _id})
361 if op_id:
362 content["op_id"] = op_id
363 self._send_msg("create", content)
364 return _id, op_id
365 except ValidationError as e:
366 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
367
368 def upload_content(self, session, _id, indata, kwargs, headers):
369 """
370 Only implemented for descriptor topics. Used for receiving content by chunks (with a transaction_id header
371 and/or gzip file. It will store and extract)
372 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
373 :param _id : the database id of entry to be updated
374 :param indata: http body request
375 :param kwargs: user query string to override parameters. NOT USED
376 :param headers: http request headers
377 :return: True package has is completely uploaded or False if partial content has been uplodaed.
378 Raise exception on error
379 """
380 raise EngineException("Method upload_content not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR)
381
382 def delete_list(self, session, filter_q=None):
383 """
384 Delete a several entries of a topic. This is for internal usage and test only, not exposed to NBI API
385 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
386 :param filter_q: filter of data to be applied
387 :return: The deleted list, it can be empty if no one match the filter.
388 """
389 # TODO add admin to filter, validate rights
390 if not filter_q:
391 filter_q = {}
392 if self.multiproject:
393 filter_q.update(self._get_project_filter(session))
394 return self.db.del_list(self.topic, filter_q)
395
396 def delete_extra(self, session, _id, db_content):
397 """
398 Delete other things apart from database entry of a item _id.
399 e.g.: other associated elements at database and other file system storage
400 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
401 :param _id: server internal id
402 :param db_content: The database content of the _id. It is already deleted when reached this method, but the
403 content is needed in same cases
404 :return: None if ok or raises EngineException with the problem
405 """
406 pass
407
408 def delete(self, session, _id, dry_run=False):
409 """
410 Delete item by its internal _id
411 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
412 :param _id: server internal id
413 :param dry_run: make checking but do not delete
414 :return: operation id (None if there is not operation), raise exception if error or not found, conflict, ...
415 """
416
417 # To allow addressing projects and users by name AS WELL AS by _id
418 filter_q = {BaseTopic.id_field(self.topic, _id): _id}
419 item_content = self.db.get_one(self.topic, filter_q)
420
421 # TODO add admin to filter, validate rights
422 # data = self.get_item(topic, _id)
423 self.check_conflict_on_del(session, _id, item_content)
424 if dry_run:
425 return None
426
427 if self.multiproject:
428 filter_q.update(self._get_project_filter(session))
429 if self.multiproject and session["project_id"]:
430 # remove reference from project_read. If not last delete
431 # if this topic is not part of session["project_id"] no midification at database is done and an exception
432 # is raised
433 self.db.set_one(self.topic, filter_q, update_dict=None,
434 pull={"_admin.projects_read": {"$in": session["project_id"]}})
435 # try to delete if there is not any more reference from projects. Ignore if it is not deleted
436 filter_q = {'_id': _id, '_admin.projects_read': [[], ["ANY"]]}
437 v = self.db.del_one(self.topic, filter_q, fail_on_empty=False)
438 if not v or not v["deleted"]:
439 return None
440 else:
441 self.db.del_one(self.topic, filter_q)
442 self.delete_extra(session, _id, item_content)
443 self._send_msg("deleted", {"_id": _id})
444 return None
445
446 def edit(self, session, _id, indata=None, kwargs=None, content=None):
447 """
448 Change the content of an item
449 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
450 :param _id: server internal id
451 :param indata: contains the changes to apply
452 :param kwargs: modifies indata
453 :param content: original content of the item
454 :return: op_id: operation id if this is processed asynchronously, None otherwise
455 """
456 indata = self._remove_envelop(indata)
457
458 # Override descriptor with query string kwargs
459 if kwargs:
460 self._update_input_with_kwargs(indata, kwargs)
461 try:
462 if indata and session.get("set_project"):
463 raise EngineException("Cannot edit content and set to project (query string SET_PROJECT) at same time",
464 HTTPStatus.UNPROCESSABLE_ENTITY)
465 indata = self._validate_input_edit(indata, force=session["force"])
466
467 # TODO self._check_edition(session, indata, _id, force)
468 if not content:
469 content = self.show(session, _id)
470 deep_update_rfc7396(content, indata)
471
472 # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
473 _id = content.get("_id") or _id
474
475 self.check_conflict_on_edit(session, content, indata, _id=_id)
476 op_id = self.format_on_edit(content, indata)
477
478 self.db.replace(self.topic, _id, content)
479
480 indata.pop("_admin", None)
481 if op_id:
482 indata["op_id"] = op_id
483 indata["_id"] = _id
484 self._send_msg("edit", indata)
485 return op_id
486 except ValidationError as e:
487 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)