bug 864 fix operation (created/edited) published to kafka

[osm/NBI.git] / osm_nbi / base_topic.py

diff --git a/osm_nbi/base_topic.py b/osm_nbi/base_topic.py
index 9688f60..70c8dff 100644 (file)
--- a/osm_nbi/base_topic.py
+++ b/osm_nbi/base_topic.py
@@ -57,6 +57,8 @@ class BaseTopic:
     schema_edit = None  # to_override
     multiproject = True  # True if this Topic can be shared by several projects. Then it contains _admin.projects_read
 
+    default_quota = 500
+
     # Alternative ID Fields for some Topics
     alt_id_field = {
         "projects": "name",
@@ -64,11 +66,12 @@ class BaseTopic:
         "roles": "name"
     }
 
-    def __init__(self, db, fs, msg):
+    def __init__(self, db, fs, msg, auth):
         self.db = db
         self.fs = fs
         self.msg = msg
         self.logger = logging.getLogger("nbi.engine")
+        self.auth = auth
 
     @staticmethod
     def id_field(topic, value):
@@ -84,6 +87,29 @@ class BaseTopic:
             return {}
         return indata
 
+    def check_quota(self, session):
+        """
+        Check whether topic quota is exceeded by the given project
+        Used by relevant topics' 'new' function to decide whether or not creation of the new item should be allowed
+        :param projects: projects (tuple) for which quota should be checked
+        :param override: boolean. If true, don't raise ValidationError even though quota be exceeded
+        :return: None
+        :raise:
+            DbException if project not found
+            ValidationError if quota exceeded and not overridden
+        """
+        if session["force"] or session["admin"]:
+            return
+        projects = session["project_id"]
+        for project in projects:
+            proj = self.auth.get_project(project)
+            pid = proj["_id"]
+            quota = proj.get("quotas", {}).get(self.topic, self.default_quota)
+            count = self.db.count(self.topic, {"_admin.projects_read": pid})
+            if count >= quota:
+                name = proj["name"]
+                raise ValidationError("{} quota ({}) exceeded for project {} ({})".format(self.topic, quota, name, pid))
+
     def _validate_input_new(self, input, force=False):
         """
         Validates input user content for a new entry. It uses jsonschema. Some overrides will use pyangbind
@@ -349,6 +375,9 @@ class BaseTopic:
              op_id: operation id if this is asynchronous, None otherwise
         """
         try:
+            if self.multiproject:
+                self.check_quota(session)
+
             content = self._remove_envelop(indata)
 
             # Override descriptor with query string kwargs
@@ -360,7 +389,7 @@ class BaseTopic:
             rollback.append({"topic": self.topic, "_id": _id})
             if op_id:
                 content["op_id"] = op_id
-            self._send_msg("create", content)
+            self._send_msg("created", content)
             return _id, op_id
         except ValidationError as e:
             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
@@ -481,7 +510,7 @@ class BaseTopic:
             if op_id:
                 indata["op_id"] = op_id
             indata["_id"] = _id
-            self._send_msg("edit", indata)
+            self._send_msg("edited", indata)
             return op_id
         except ValidationError as e:
             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)