X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FNBI.git;a=blobdiff_plain;f=osm_nbi%2Fengine.py;h=f186a4c34caade844f5f8835e141f2c8b16d506d;hp=4713b89f698c557273756ecd76766fe211ed832a;hb=e4c83a7871f341a8340e6befc9b4e66a72e4927d;hpb=07f3172ef24dab2c7c47c9a8e726f1f9876f13b3

diff --git a/osm_nbi/engine.py b/osm_nbi/engine.py
index 4713b89..f186a4c 100644
--- a/osm_nbi/engine.py
+++ b/osm_nbi/engine.py
@@ -1,5 +1,18 @@
 # -*- coding: utf-8 -*-
 
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import logging
 from osm_common import dbmongo, dbmemory, fslocal, msglocal, msgkafka, version as common_version
 from osm_common.dbbase import DbException
@@ -7,11 +20,12 @@ from osm_common.fsbase import FsException
 from osm_common.msgbase import MsgException
 from http import HTTPStatus
 from base_topic import EngineException, versiontuple
-from admin_topics import UserTopic, ProjectTopic, VimAccountTopic, SdnTopic
+from admin_topics import UserTopic, ProjectTopic, VimAccountTopic, WimAccountTopic, SdnTopic
 from descriptor_topics import VnfdTopic, NsdTopic, PduTopic, NstTopic
 from instance_topics import NsrTopic, VnfrTopic, NsLcmOpTopic, NsiTopic, NsiLcmOpTopic
 from base64 import b64encode
 from os import urandom
+from threading import Lock
 
 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
 min_common_version = "0.1.8"
@@ -27,6 +41,7 @@ class Engine(object):
         "vnfrs": VnfrTopic,
         "nslcmops": NsLcmOpTopic,
         "vim_accounts": VimAccountTopic,
+        "wim_accounts": WimAccountTopic,
         "sdns": SdnTopic,
         "users": UserTopic,
         "projects": ProjectTopic,
@@ -42,6 +57,7 @@ class Engine(object):
         self.config = None
         self.logger = logging.getLogger("nbi.engine")
         self.map_topic = {}
+        self.write_lock = None
 
     def start(self, config):
         """
@@ -84,6 +100,7 @@ class Engine(object):
                     raise EngineException("Invalid configuration param '{}' at '[message]':'driver'".format(
                         config["storage"]["driver"]))
 
+            self.write_lock = Lock()
             # create one class per topic
             for topic, topic_class in self.map_from_topic_to_class.items():
                 self.map_topic[topic] = topic_class(self.db, self.fs, self.msg)
@@ -98,6 +115,7 @@ class Engine(object):
                 self.fs.fs_disconnect()
             if self.fs:
                 self.fs.fs_disconnect()
+            self.write_lock = None
         except (DbException, FsException, MsgException) as e:
             raise EngineException(str(e), http_code=e.http_code)
 
@@ -116,7 +134,8 @@ class Engine(object):
         """
         if topic not in self.map_topic:
             raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
-        return self.map_topic[topic].new(rollback, session, indata, kwargs, headers, force)
+        with self.write_lock:
+            return self.map_topic[topic].new(rollback, session, indata, kwargs, headers, force)
 
     def upload_content(self, session, topic, _id, indata, kwargs, headers, force=False):
         """
@@ -132,7 +151,8 @@ class Engine(object):
         """
         if topic not in self.map_topic:
             raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
-        return self.map_topic[topic].upload_content(session, _id, indata, kwargs, headers, force)
+        with self.write_lock:
+            return self.map_topic[topic].upload_content(session, _id, indata, kwargs, headers, force)
 
     def get_item_list(self, session, topic, filter_q=None):
         """
@@ -182,7 +202,8 @@ class Engine(object):
         """
         if topic not in self.map_topic:
             raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
-        return self.map_topic[topic].delete_list(session, _filter)
+        with self.write_lock:
+            return self.map_topic[topic].delete_list(session, _filter)
 
     def del_item(self, session, topic, _id, force=False):
         """
@@ -195,7 +216,8 @@ class Engine(object):
         """
         if topic not in self.map_topic:
             raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
-        return self.map_topic[topic].delete(session, _id, force)
+        with self.write_lock:
+            return self.map_topic[topic].delete(session, _id, force)
 
     def edit_item(self, session, topic, _id, indata=None, kwargs=None, force=False):
         """
@@ -210,14 +232,8 @@ class Engine(object):
         """
         if topic not in self.map_topic:
             raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
-        return self.map_topic[topic].edit(session, _id, indata, kwargs, force)
-
-    def prune(self):
-        """
-        Prune database not needed content
-        :return: None
-        """
-        return self.db.del_list("nsrs", {"_admin.to_delete": True})
+        with self.write_lock:
+            return self.map_topic[topic].edit(session, _id, indata, kwargs, force)
 
     def create_admin(self):
         """
@@ -252,6 +268,7 @@ class Engine(object):
                 }
                 self.db.create("admin", version_data)
                 self.db.set_secret_key(serial)
+                return
             # TODO add future migrations here
 
         raise EngineException("Wrong database version '{}'. Expected '{}'"