fix(api): multi-line search

veops · Dec 23, 2024 · b253fdf · b253fdf
1 parent d012943
commit b253fdf
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 13 deletions.
diff --git a/cmdb-api/api/lib/cmdb/ci_type.py b/cmdb-api/api/lib/cmdb/ci_type.py
@@ -1145,13 +1145,14 @@ def get_by_type_id(type_id, need_other=False):
             else:
                 group_pos = group2pos[group['name']]
 
-            attr = None
             for i in items:
                 if i.attr_id in id2attr:
                     attr = id2attr[i.attr_id]
                     attr['inherited'] = group['inherited']
                     attr['inherited_from'] = group.get('inherited_from')
                     result[group_pos]['attributes'].append(attr)
+                else:
+                    continue
 
                 if i.attr_id in attr2pos:
                     result[attr2pos[i.attr_id][0]]['attributes'].remove(attr2pos[i.attr_id][1])

diff --git a/cmdb-api/api/lib/cmdb/ipam/history.py b/cmdb-api/api/lib/cmdb/ipam/history.py
@@ -43,7 +43,7 @@ def add(self, **kwargs):
 
         if kwargs.get('ips'):
             from api.lib.cmdb.ipam.address import IpAddressManager
-            IpAddressManager().assign_ips(kwargs['ips'], None, kwargs.get('cidr'),
+            IpAddressManager().assign_ips(kwargs['ips'], ci_id, kwargs.get('cidr'),
                                           **{IPAddressBuiltinAttributes.IS_USED: 1})
 
             scan_rule = IPAMSubnetScan.get_by(ci_id=ci_id, first=True, to_dict=False)

diff --git a/cmdb-api/api/lib/cmdb/search/ci/db/search.py b/cmdb-api/api/lib/cmdb/search/ci/db/search.py
@@ -142,7 +142,7 @@ def _type_query_handler(self, v, queries, is_sub=False):
                                     if str(ci_type.id) in self.type_id_list:
                                         self.type_id_list.remove(str(ci_type.id))
                                     type_id_list.remove(str(ci_type.id))
-                                    sub.extend([i for i in queries[1:] if isinstance(i, six.string_types)])
+                                    sub.extend([i for i in queries[1:] if isinstance(i, (six.string_types, list))])
 
                                     sub.insert(0, "_type:{}".format(ci_type.id))
                                     queries.append(dict(operator="|", queries=sub))
@@ -434,11 +434,14 @@ def __confirm_type_first(self, queries):
                 if not q.startswith("("):
                     raise SearchError(ErrFormat.ci_search_Parentheses_invalid)
 
-                operator, q = self._operator_proc(q)
-                if q.endswith(")"):
-                    result.append(dict(operator=operator, queries=[q[1:-1]]))
+                if ":" not in q:  # multi-line search
+                    result.append(q[1:-1].split(';'))
+                else:
+                    operator, q = self._operator_proc(q)
+                    if q.endswith(")"):
+                        result.append(dict(operator=operator, queries=[q[1:-1]]))
 
-                sub = dict(operator=operator, queries=[q[1:]])
+                    sub = dict(operator=operator, queries=[q[1:]])
             elif q.endswith(")") and sub:
                 sub['queries'].append(q[:-1])
                 result.append(copy.deepcopy(sub))
@@ -526,27 +529,31 @@ def __query_build_by_field(self, queries, is_first=True, only_type_query_special
         query_sql = ""
 
         for q in queries:
+            # current_app.logger.debug(q)
             _query_sql = ""
             if isinstance(q, dict):
-                current_app.logger.debug("Dict query content: queries=%s, operator=%s", q['queries'], q['operator'])
                 if len(q['queries']) == 1 and ";" in q['queries'][0]:
                     values = q['queries'][0].split(";")
                     in_values = ",".join("'{0}'".format(v) for v in values)
                     _query_sql = QUERY_CI_BY_NO_ATTR_IN.format(in_values, alias)
                     operator = q['operator']
                 else:
-                    alias, _query_sql, operator = self.__query_build_by_field(q['queries'], True, True, alias, is_sub=True)
+                    alias, _query_sql, operator = self.__query_build_by_field(q['queries'], True, True, alias,
+                                                                              is_sub=True)
                     operator = q['operator']
 
             elif ":" in q and not q.startswith("*"):
                 alias, _query_sql, operator = self.__query_by_attr(q, queries, alias, is_sub)
             elif q == "*":
                 continue
             elif q:
-                q = q.replace("'", "\\'")
-                q = q.replace('"', '\\"')
-                q = q.replace("*", "%").replace('\\n', '%')
-                _query_sql = QUERY_CI_BY_NO_ATTR.format(q, alias)
+                if not isinstance(q, list):
+                    q = q.replace("'", "\\'")
+                    q = q.replace('"', '\\"')
+                    q = q.replace("*", "%").replace('\\n', '%')
+                    _query_sql = QUERY_CI_BY_NO_ATTR.format(q, alias)
+                else:
+                    _query_sql = QUERY_CI_BY_NO_ATTR_IN.format(",".join("'{0}'".format(v) for v in q), alias)
 
             if is_first and _query_sql and not self.only_type_query:
                 query_sql = "SELECT * FROM ({0}) AS {1}".format(_query_sql, alias)