7x31
/
cmdb
mirror of https://github.com/veops/cmdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

pref(api): A CIType allows repeated binding of auto-discovery rules (#313)

This commit is contained in:
pycook 2023-12-16 17:56:14 +08:00 committed by GitHub
parent b45fd0cbbb
commit 092a8b9b92
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 12 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
cmdb-api/api/lib/cmdb/auto_discovery/auto_discovery.py
View File

@ -3,11 +3,6 @@ import datetime
import json
import os
from flask import abort
from flask import current_app
from flask_login import current_user
from sqlalchemy import func
from api.extensions import db
from api.lib.cmdb.auto_discovery.const import ClOUD_MAP
from api.lib.cmdb.cache import CITypeAttributeCache
@ -28,6 +23,10 @@ from api.lib.utils import AESCrypto
from api.models.cmdb import AutoDiscoveryCI
from api.models.cmdb import AutoDiscoveryCIType
from api.models.cmdb import AutoDiscoveryRule
from flask import abort
from flask import current_app
from flask_login import current_user
from sqlalchemy import func
PWD = os.path.abspath(os.path.dirname(__file__))
@ -251,20 +250,17 @@ class AutoDiscoveryCITypeCRUD(DBMixin):
current_app.logger.warning(e)
return abort(400, str(e))
def _can_add(self, **kwargs):
self.cls.get_by(type_id=kwargs['type_id'], adr_id=kwargs.get('adr_id') or None) and abort(
400, ErrFormat.ad_duplicate)
# self.__valid_exec_target(kwargs.get('agent_id'), kwargs.get('query_expr'))
@staticmethod
def _can_add(**kwargs):
if kwargs.get('adr_id'):
adr = AutoDiscoveryRule.get_by_id(kwargs['adr_id']) or abort(
AutoDiscoveryRule.get_by_id(kwargs['adr_id']) or abort(
404, ErrFormat.adr_not_found.format("id={}".format(kwargs['adr_id'])))
if not adr.is_plugin:
other = self.cls.get_by(adr_id=adr.id, first=True, to_dict=False)
if other:
ci_type = CITypeCache.get(other.type_id)
return abort(400, ErrFormat.adr_default_ref_once.format(ci_type.alias))
# if not adr.is_plugin:
# other = self.cls.get_by(adr_id=adr.id, first=True, to_dict=False)
# if other:
# ci_type = CITypeCache.get(other.type_id)
# return abort(400, ErrFormat.adr_default_ref_once.format(ci_type.alias))
if kwargs.get('is_plugin') and kwargs.get('plugin_script'):
kwargs = check_plugin_script(**kwargs)