Coverage for osm_nbi/base_topic.py: 78%
326 statements
« prev ^ index » next coverage.py v7.3.1, created at 2024-06-25 09:03 +0000
1# -*- coding: utf-8 -*-
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.
16import logging
17from uuid import uuid4
18from http import HTTPStatus
19from time import time
20from osm_common.dbbase import deep_update_rfc7396, DbException
21from osm_nbi.validation import validate_input, ValidationError, is_valid_uuid
22from yaml import safe_load, YAMLError
24__author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
27class EngineException(Exception):
28 def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
29 self.http_code = http_code
30 super(Exception, self).__init__(message)
33class NBIBadArgumentsException(Exception):
34 """
35 Bad argument values exception
36 """
38 def __init__(self, message: str = "", bad_args: list = None):
39 Exception.__init__(self, message)
40 self.message = message
41 self.bad_args = bad_args
43 def __str__(self):
44 return "{}, Bad arguments: {}".format(self.message, self.bad_args)
47def deep_get(target_dict, key_list):
48 """
49 Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
50 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
51 :param target_dict: dictionary to be read
52 :param key_list: list of keys to read from target_dict
53 :return: The wanted value if exist, None otherwise
54 """
55 for key in key_list:
56 if not isinstance(target_dict, dict) or key not in target_dict:
57 return None
58 target_dict = target_dict[key]
59 return target_dict
62def detect_descriptor_usage(descriptor: dict, db_collection: str, db: object) -> bool:
63 """Detect the descriptor usage state.
65 Args:
66 descriptor (dict): VNF or NS Descriptor as dictionary
67 db_collection (str): collection name which is looked for in DB
68 db (object): name of db object
70 Returns:
71 True if descriptor is in use else None
73 """
74 try:
75 if not descriptor:
76 raise NBIBadArgumentsException(
77 "Argument is mandatory and can not be empty", "descriptor"
78 )
80 if not db:
81 raise NBIBadArgumentsException("A valid DB object should be provided", "db")
83 search_dict = {
84 "vnfds": ("vnfrs", "vnfd-id"),
85 "nsds": ("nsrs", "nsd-id"),
86 }
88 if db_collection not in search_dict:
89 raise NBIBadArgumentsException(
90 "db_collection should be equal to vnfds or nsds", "db_collection"
91 )
93 record_list = db.get_list(
94 search_dict[db_collection][0],
95 {search_dict[db_collection][1]: descriptor["_id"]},
96 )
98 if record_list:
99 return True
101 except (DbException, KeyError, NBIBadArgumentsException) as error:
102 raise EngineException(
103 f"Error occured while detecting the descriptor usage: {error}"
104 )
107def update_descriptor_usage_state(
108 descriptor: dict, db_collection: str, db: object
109) -> None:
110 """Updates the descriptor usage state.
112 Args:
113 descriptor (dict): VNF or NS Descriptor as dictionary
114 db_collection (str): collection name which is looked for in DB
115 db (object): name of db object
117 Returns:
118 None
120 """
121 try:
122 descriptor_update = {
123 "_admin.usageState": "NOT_IN_USE",
124 }
126 if detect_descriptor_usage(descriptor, db_collection, db):
127 descriptor_update = {
128 "_admin.usageState": "IN_USE",
129 }
131 db.set_one(
132 db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
133 )
135 except (DbException, KeyError, NBIBadArgumentsException) as error:
136 raise EngineException(
137 f"Error occured while updating the descriptor usage state: {error}"
138 )
141def get_iterable(input_var):
142 """
143 Returns an iterable, in case input_var is None it just returns an empty tuple
144 :param input_var: can be a list, tuple or None
145 :return: input_var or () if it is None
146 """
147 if input_var is None:
148 return ()
149 return input_var
152def versiontuple(v):
153 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
154 filled = []
155 for point in v.split("."):
156 filled.append(point.zfill(8))
157 return tuple(filled)
160def increment_ip_mac(ip_mac, vm_index=1):
161 if not isinstance(ip_mac, str):
162 return ip_mac
163 try:
164 # try with ipv4 look for last dot
165 i = ip_mac.rfind(".")
166 if i > 0:
167 i += 1
168 return "{}{}".format(ip_mac[:i], int(ip_mac[i:]) + vm_index)
169 # try with ipv6 or mac look for last colon. Operate in hex
170 i = ip_mac.rfind(":")
171 if i > 0:
172 i += 1
173 # format in hex, len can be 2 for mac or 4 for ipv6
174 return ("{}{:0" + str(len(ip_mac) - i) + "x}").format(
175 ip_mac[:i], int(ip_mac[i:], 16) + vm_index
176 )
177 except Exception:
178 pass
179 return None
182class BaseTopic:
183 # static variables for all instance classes
184 topic = None # to_override
185 topic_msg = None # to_override
186 quota_name = None # to_override. If not provided topic will be used for quota_name
187 schema_new = None # to_override
188 schema_edit = None # to_override
189 multiproject = True # True if this Topic can be shared by several projects. Then it contains _admin.projects_read
191 default_quota = 500
193 # Alternative ID Fields for some Topics
194 alt_id_field = {"projects": "name", "users": "username", "roles": "name"}
196 def __init__(self, db, fs, msg, auth):
197 self.db = db
198 self.fs = fs
199 self.msg = msg
200 self.logger = logging.getLogger("nbi.engine")
201 self.auth = auth
203 @staticmethod
204 def id_field(topic, value):
205 """Returns ID Field for given topic and field value"""
206 if topic in BaseTopic.alt_id_field.keys() and not is_valid_uuid(value):
207 return BaseTopic.alt_id_field[topic]
208 else:
209 return "_id"
211 @staticmethod
212 def _remove_envelop(indata=None):
213 if not indata:
214 return {}
215 return indata
217 def check_quota(self, session):
218 """
219 Check whether topic quota is exceeded by the given project
220 Used by relevant topics' 'new' function to decide whether or not creation of the new item should be allowed
221 :param session[project_id]: projects (tuple) for which quota should be checked
222 :param session[force]: boolean. If true, skip quota checking
223 :return: None
224 :raise:
225 DbException if project not found
226 ValidationError if quota exceeded in one of the projects
227 """
228 if session["force"]:
229 return
230 projects = session["project_id"]
231 for project in projects:
232 proj = self.auth.get_project(project)
233 pid = proj["_id"]
234 quota_name = self.quota_name or self.topic
235 quota = proj.get("quotas", {}).get(quota_name, self.default_quota)
236 count = self.db.count(self.topic, {"_admin.projects_read": pid})
237 if count >= quota:
238 name = proj["name"]
239 raise ValidationError(
240 "quota ({}={}) exceeded for project {} ({})".format(
241 quota_name, quota, name, pid
242 ),
243 http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
244 )
246 def _validate_input_new(self, input, force=False):
247 """
248 Validates input user content for a new entry. It uses jsonschema. Some overrides will use pyangbind
249 :param input: user input content for the new topic
250 :param force: may be used for being more tolerant
251 :return: The same input content, or a changed version of it.
252 """
253 if self.schema_new:
254 validate_input(input, self.schema_new)
255 return input
257 def _validate_input_edit(self, input, content, force=False):
258 """
259 Validates input user content for an edition. It uses jsonschema. Some overrides will use pyangbind
260 :param input: user input content for the new topic
261 :param force: may be used for being more tolerant
262 :return: The same input content, or a changed version of it.
263 """
264 if self.schema_edit:
265 validate_input(input, self.schema_edit)
266 return input
268 @staticmethod
269 def _get_project_filter(session):
270 """
271 Generates a filter dictionary for querying database, so that only allowed items for this project can be
272 addressed. Only proprietary or public can be used. Allowed projects are at _admin.project_read/write. If it is
273 not present or contains ANY mean public.
274 :param session: contains:
275 project_id: project list this session has rights to access. Can be empty, one or several
276 set_project: items created will contain this project list
277 force: True or False
278 public: True, False or None
279 method: "list", "show", "write", "delete"
280 admin: True or False
281 :return: dictionary with project filter
282 """
283 p_filter = {}
284 project_filter_n = []
285 project_filter = list(session["project_id"])
287 if session["method"] not in ("list", "delete"):
288 if project_filter:
289 project_filter.append("ANY")
290 elif session["public"] is not None:
291 if session["public"]:
292 project_filter.append("ANY")
293 else:
294 project_filter_n.append("ANY")
296 if session.get("PROJECT.ne"):
297 project_filter_n.append(session["PROJECT.ne"])
299 if project_filter:
300 if session["method"] in ("list", "show", "delete") or session.get(
301 "set_project"
302 ):
303 p_filter["_admin.projects_read.cont"] = project_filter
304 else:
305 p_filter["_admin.projects_write.cont"] = project_filter
306 if project_filter_n:
307 if session["method"] in ("list", "show", "delete") or session.get(
308 "set_project"
309 ):
310 p_filter["_admin.projects_read.ncont"] = project_filter_n
311 else:
312 p_filter["_admin.projects_write.ncont"] = project_filter_n
314 return p_filter
316 def check_conflict_on_new(self, session, indata):
317 """
318 Check that the data to be inserted is valid
319 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
320 :param indata: data to be inserted
321 :return: None or raises EngineException
322 """
323 pass
325 def check_conflict_on_edit(self, session, final_content, edit_content, _id):
326 """
327 Check that the data to be edited/uploaded is valid
328 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
329 :param final_content: data once modified. This method may change it.
330 :param edit_content: incremental data that contains the modifications to apply
331 :param _id: internal _id
332 :return: final_content or raises EngineException
333 """
334 if not self.multiproject:
335 return final_content
336 # Change public status
337 if session["public"] is not None:
338 if (
339 session["public"]
340 and "ANY" not in final_content["_admin"]["projects_read"]
341 ):
342 final_content["_admin"]["projects_read"].append("ANY")
343 final_content["_admin"]["projects_write"].clear()
344 if (
345 not session["public"]
346 and "ANY" in final_content["_admin"]["projects_read"]
347 ):
348 final_content["_admin"]["projects_read"].remove("ANY")
350 # Change project status
351 if session.get("set_project"):
352 for p in session["set_project"]:
353 if p not in final_content["_admin"]["projects_read"]:
354 final_content["_admin"]["projects_read"].append(p)
356 return final_content
358 def check_unique_name(self, session, name, _id=None):
359 """
360 Check that the name is unique for this project
361 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
362 :param name: name to be checked
363 :param _id: If not None, ignore this entry that are going to change
364 :return: None or raises EngineException
365 """
366 if not self.multiproject:
367 _filter = {}
368 else:
369 _filter = self._get_project_filter(session)
370 _filter["name"] = name
371 if _id:
372 _filter["_id.neq"] = _id
373 if self.db.get_one(
374 self.topic, _filter, fail_on_empty=False, fail_on_more=False
375 ):
376 raise EngineException(
377 "name '{}' already exists for {}".format(name, self.topic),
378 HTTPStatus.CONFLICT,
379 )
381 @staticmethod
382 def format_on_new(content, project_id=None, make_public=False):
383 """
384 Modifies content descriptor to include _admin
385 :param content: descriptor to be modified
386 :param project_id: if included, it add project read/write permissions. Can be None or a list
387 :param make_public: if included it is generated as public for reading.
388 :return: op_id: operation id on asynchronous operation, None otherwise. In addition content is modified
389 """
390 now = time()
391 if "_admin" not in content:
392 content["_admin"] = {}
393 if not content["_admin"].get("created"):
394 content["_admin"]["created"] = now
395 content["_admin"]["modified"] = now
396 if not content.get("_id"):
397 content["_id"] = str(uuid4())
398 if project_id is not None:
399 if not content["_admin"].get("projects_read"):
400 content["_admin"]["projects_read"] = list(project_id)
401 if make_public:
402 content["_admin"]["projects_read"].append("ANY")
403 if not content["_admin"].get("projects_write"):
404 content["_admin"]["projects_write"] = list(project_id)
405 return None
407 @staticmethod
408 def format_on_edit(final_content, edit_content):
409 """
410 Modifies final_content to admin information upon edition
411 :param final_content: final content to be stored at database
412 :param edit_content: user requested update content
413 :return: operation id, if this edit implies an asynchronous operation; None otherwise
414 """
415 if final_content.get("_admin"):
416 now = time()
417 final_content["_admin"]["modified"] = now
418 return None
420 def _send_msg(self, action, content, not_send_msg=None):
421 if self.topic_msg and not_send_msg is not False:
422 content = content.copy()
423 content.pop("_admin", None)
424 if isinstance(not_send_msg, list):
425 not_send_msg.append((self.topic_msg, action, content))
426 else:
427 self.msg.write(self.topic_msg, action, content)
429 def check_conflict_on_del(self, session, _id, db_content):
430 """
431 Check if deletion can be done because of dependencies if it is not force. To override
432 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
433 :param _id: internal _id
434 :param db_content: The database content of this item _id
435 :return: None if ok or raises EngineException with the conflict
436 """
437 pass
439 @staticmethod
440 def _update_input_with_kwargs(desc, kwargs, yaml_format=False):
441 """
442 Update descriptor with the kwargs. It contains dot separated keys
443 :param desc: dictionary to be updated
444 :param kwargs: plain dictionary to be used for updating.
445 :param yaml_format: get kwargs values as yaml format.
446 :return: None, 'desc' is modified. It raises EngineException.
447 """
448 if not kwargs:
449 return
450 try:
451 for k, v in kwargs.items():
452 update_content = desc
453 kitem_old = None
454 klist = k.split(".")
455 for kitem in klist:
456 if kitem_old is not None:
457 update_content = update_content[kitem_old]
458 if isinstance(update_content, dict):
459 kitem_old = kitem
460 if not isinstance(update_content.get(kitem_old), (dict, list)):
461 update_content[kitem_old] = {}
462 elif isinstance(update_content, list):
463 # key must be an index of the list, must be integer
464 kitem_old = int(kitem)
465 # if index greater than list, extend the list
466 if kitem_old >= len(update_content):
467 update_content += [None] * (
468 kitem_old - len(update_content) + 1
469 )
470 if not isinstance(update_content[kitem_old], (dict, list)):
471 update_content[kitem_old] = {}
472 else:
473 raise EngineException(
474 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(
475 k, kitem
476 )
477 )
478 if v is None:
479 del update_content[kitem_old]
480 else:
481 update_content[kitem_old] = v if not yaml_format else safe_load(v)
482 except KeyError:
483 raise EngineException(
484 "Invalid query string '{}'. Descriptor does not contain '{}'".format(
485 k, kitem_old
486 )
487 )
488 except ValueError:
489 raise EngineException(
490 "Invalid query string '{}'. Expected integer index list instead of '{}'".format(
491 k, kitem
492 )
493 )
494 except IndexError:
495 raise EngineException(
496 "Invalid query string '{}'. Index '{}' out of range".format(
497 k, kitem_old
498 )
499 )
500 except YAMLError:
501 raise EngineException("Invalid query string '{}' yaml format".format(k))
503 def sol005_projection(self, data):
504 # Projection was moved to child classes
505 return data
507 def show(self, session, _id, filter_q=None, api_req=False):
508 """
509 Get complete information on an topic
510 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
511 :param _id: server internal id
512 :param filter_q: dict: query parameter
513 :param api_req: True if this call is serving an external API request. False if serving internal request.
514 :return: dictionary, raise exception if not found.
515 """
516 if not self.multiproject:
517 filter_db = {}
518 else:
519 filter_db = self._get_project_filter(session)
520 # To allow project&user addressing by name AS WELL AS _id
521 filter_db[BaseTopic.id_field(self.topic, _id)] = _id
522 data = self.db.get_one(self.topic, filter_db)
524 # Only perform SOL005 projection if we are serving an external request
525 if api_req:
526 self.sol005_projection(data)
528 return data
530 # TODO transform data for SOL005 URL requests
531 # TODO remove _admin if not admin
533 def get_file(self, session, _id, path=None, accept_header=None):
534 """
535 Only implemented for descriptor topics. Return the file content of a descriptor
536 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
537 :param _id: Identity of the item to get content
538 :param path: artifact path or "$DESCRIPTOR" or None
539 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
540 :return: opened file or raises an exception
541 """
542 raise EngineException(
543 "Method get_file not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR
544 )
546 def list(self, session, filter_q=None, api_req=False):
547 """
548 Get a list of the topic that matches a filter
549 :param session: contains the used login username and working project
550 :param filter_q: filter of data to be applied
551 :param api_req: True if this call is serving an external API request. False if serving internal request.
552 :return: The list, it can be empty if no one match the filter.
553 """
554 if not filter_q:
555 filter_q = {}
556 if self.multiproject:
557 filter_q.update(self._get_project_filter(session))
559 # TODO transform data for SOL005 URL requests. Transform filtering
560 # TODO implement "field-type" query string SOL005
561 data = self.db.get_list(self.topic, filter_q)
563 # Only perform SOL005 projection if we are serving an external request
564 if api_req:
565 data = [self.sol005_projection(inst) for inst in data]
567 return data
569 def new(self, rollback, session, indata=None, kwargs=None, headers=None):
570 """
571 Creates a new entry into database.
572 :param rollback: list to append created items at database in case a rollback may to be done
573 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
574 :param indata: data to be inserted
575 :param kwargs: used to override the indata descriptor
576 :param headers: http request headers
577 :return: _id, op_id:
578 _id: identity of the inserted data.
579 op_id: operation id if this is asynchronous, None otherwise
580 """
581 try:
582 if self.multiproject:
583 self.check_quota(session)
585 content = self._remove_envelop(indata)
587 # Override descriptor with query string kwargs
588 self._update_input_with_kwargs(content, kwargs)
589 content = self._validate_input_new(content, force=session["force"])
590 self.check_conflict_on_new(session, content)
591 op_id = self.format_on_new(
592 content, project_id=session["project_id"], make_public=session["public"]
593 )
594 _id = self.db.create(self.topic, content)
595 rollback.append({"topic": self.topic, "_id": _id})
596 if op_id:
597 content["op_id"] = op_id
598 self._send_msg("created", content)
599 return _id, op_id
600 except ValidationError as e:
601 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
603 def upload_content(self, session, _id, indata, kwargs, headers):
604 """
605 Only implemented for descriptor topics. Used for receiving content by chunks (with a transaction_id header
606 and/or gzip file. It will store and extract)
607 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
608 :param _id : the database id of entry to be updated
609 :param indata: http body request
610 :param kwargs: user query string to override parameters. NOT USED
611 :param headers: http request headers
612 :return: True package has is completely uploaded or False if partial content has been uplodaed.
613 Raise exception on error
614 """
615 raise EngineException(
616 "Method upload_content not valid for this topic",
617 HTTPStatus.INTERNAL_SERVER_ERROR,
618 )
620 def delete_list(self, session, filter_q=None):
621 """
622 Delete a several entries of a topic. This is for internal usage and test only, not exposed to NBI API
623 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
624 :param filter_q: filter of data to be applied
625 :return: The deleted list, it can be empty if no one match the filter.
626 """
627 # TODO add admin to filter, validate rights
628 if not filter_q:
629 filter_q = {}
630 if self.multiproject:
631 filter_q.update(self._get_project_filter(session))
632 return self.db.del_list(self.topic, filter_q)
634 def delete_extra(self, session, _id, db_content, not_send_msg=None):
635 """
636 Delete other things apart from database entry of a item _id.
637 e.g.: other associated elements at database and other file system storage
638 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
639 :param _id: server internal id
640 :param db_content: The database content of the _id. It is already deleted when reached this method, but the
641 content is needed in same cases
642 :param not_send_msg: To not send message (False) or store content (list) instead
643 :return: None if ok or raises EngineException with the problem
644 """
645 pass
647 def delete(self, session, _id, dry_run=False, not_send_msg=None):
648 """
649 Delete item by its internal _id
650 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
651 :param _id: server internal id
652 :param dry_run: make checking but do not delete
653 :param not_send_msg: To not send message (False) or store content (list) instead
654 :return: operation id (None if there is not operation), raise exception if error or not found, conflict, ...
655 """
657 # To allow addressing projects and users by name AS WELL AS by _id
658 if not self.multiproject:
659 filter_q = {}
660 else:
661 filter_q = self._get_project_filter(session)
662 filter_q[self.id_field(self.topic, _id)] = _id
663 item_content = self.db.get_one(self.topic, filter_q)
665 self.check_conflict_on_del(session, _id, item_content)
666 if dry_run:
667 return None
669 if self.multiproject and session["project_id"]:
670 # remove reference from project_read if there are more projects referencing it. If it last one,
671 # do not remove reference, but delete
672 other_projects_referencing = next(
673 (
674 p
675 for p in item_content["_admin"]["projects_read"]
676 if p not in session["project_id"] and p != "ANY"
677 ),
678 None,
679 )
681 # check if there are projects referencing it (apart from ANY, that means, public)....
682 if other_projects_referencing:
683 # remove references but not delete
684 update_dict_pull = {
685 "_admin.projects_read": session["project_id"],
686 "_admin.projects_write": session["project_id"],
687 }
688 self.db.set_one(
689 self.topic, filter_q, update_dict=None, pull_list=update_dict_pull
690 )
691 return None
692 else:
693 can_write = next(
694 (
695 p
696 for p in item_content["_admin"]["projects_write"]
697 if p == "ANY" or p in session["project_id"]
698 ),
699 None,
700 )
701 if not can_write:
702 raise EngineException(
703 "You have not write permission to delete it",
704 http_code=HTTPStatus.UNAUTHORIZED,
705 )
707 # delete
708 self.db.del_one(self.topic, filter_q)
709 self.delete_extra(session, _id, item_content, not_send_msg=not_send_msg)
710 self._send_msg("deleted", {"_id": _id}, not_send_msg=not_send_msg)
711 return None
713 def edit(self, session, _id, indata=None, kwargs=None, content=None):
714 """
715 Change the content of an item
716 :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
717 :param _id: server internal id
718 :param indata: contains the changes to apply
719 :param kwargs: modifies indata
720 :param content: original content of the item
721 :return: op_id: operation id if this is processed asynchronously, None otherwise
722 """
723 indata = self._remove_envelop(indata)
725 # Override descriptor with query string kwargs
726 if kwargs:
727 self._update_input_with_kwargs(indata, kwargs)
728 try:
729 if indata and session.get("set_project"):
730 raise EngineException(
731 "Cannot edit content and set to project (query string SET_PROJECT) at same time",
732 HTTPStatus.UNPROCESSABLE_ENTITY,
733 )
734 # TODO self._check_edition(session, indata, _id, force)
735 if not content:
736 content = self.show(session, _id)
737 indata = self._validate_input_edit(indata, content, force=session["force"])
738 deep_update_rfc7396(content, indata)
740 # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
741 _id = content.get("_id") or _id
743 content = self.check_conflict_on_edit(session, content, indata, _id=_id)
744 op_id = self.format_on_edit(content, indata)
746 self.db.replace(self.topic, _id, content)
748 indata.pop("_admin", None)
749 if op_id:
750 indata["op_id"] = op_id
751 indata["_id"] = _id
752 self._send_msg("edited", indata)
753 return op_id
754 except ValidationError as e:
755 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)