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