Fix the IPv6 does not work
[pti/o2.git] / o2common / views / view.py
1 # Copyright (C) 2021-2022 Wind River Systems, Inc.
2 #
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 implied.
12 #  See the License for the specific language governing permissions and
13 #  limitations under the License.
14
15 # import re
16 from sqlalchemy.sql.elements import ColumnElement
17
18 from o2common.views.route_exception import BadRequestException
19 from o2common.domain.filter import gen_orm_filter, \
20     transfer_filter_attr_name_in_special
21
22 from o2common.helper import o2logging
23 logger = o2logging.get_logger(__name__)
24
25
26 def gen_filter(obj: ColumnElement, filter_str: str):
27     check_filter(obj, filter_str)
28     try:
29         filter_list = gen_orm_filter(obj, filter_str)
30     except KeyError as e:
31         raise BadRequestException(e.args[0])
32     return filter_list
33
34
35 # The regular expressions testing example put on here
36 # (neq,testkey,value-1)
37 # (neq,testkey,value-1,value-2)
38 # (gt,hello,1)
39 # (gte,world,2)
40 # (lt,testlt,notint)
41 # (ncont,key1,v1,v_2)
42 # (gt,hello,1);(ncont,world,val1,val-2)
43 # (eq,wrong,60cba7be-e2cd-3b8c-a7ff-16e0f10573f9)
44 # (eq,description,value key)
45 def check_filter(obj: ColumnElement, filter_str: str):
46     if not filter_str:
47         return
48     # pattern = r'^(\((eq|neq|gt|lt|gte|lte){1},\w+,[\w -\.]+\)\;?|' +\
49     #     r'\((in|nin|cont|ncont){1},\w*(,[\w -\.]*)*\)\;?)+'
50     # result = re.match(pattern, filter_str)
51     # logger.debug('filter: {} match result is {}'.format(filter_str, result))
52     # if not result:
53     #     raise BadRequestException(
54     #         'filter value format is invalid')
55     check_filter_attribute(obj, filter_str)
56
57
58 def check_filter_attribute(obj: ColumnElement, filter_str: str):
59     filter_without_space = filter_str.strip()
60     logger.debug(
61         f"filter_str: {filter_str}, stripped: {filter_without_space}")
62     items = filter_without_space.split(';')
63
64     for i in items:
65         item = i.strip(' ()')
66         filter_expr = item.split(',')
67         if len(filter_expr) < 3:
68             raise BadRequestException(
69                 'invalid filter {}'.format(i))
70         filter_op = filter_expr[0].strip()
71         filter_key = filter_expr[1].strip()
72         if filter_key == 'objectType':
73             logger.debug('ignore objectType while checking formatter')
74             continue
75         filter_vals = filter_expr[2:]
76         if filter_op in ["eq", "neq", "gt", "lt", "gte", "lte"]:
77             if len(filter_vals) != 1:
78                 raise BadRequestException(
79                     "Found {} values: {} while only single value"
80                     " is allowed for operation {}".format(
81                         len(filter_vals), filter_vals, filter_op)
82                 )
83         elif filter_op not in ["in", "nin", "cont", "ncont"]:
84             raise BadRequestException(
85                 'Filter operation {} is invalid'.format(filter_op)
86             )
87         else:
88             pass
89         filter_key = transfer_filter_attr_name_in_special(obj, filter_key)
90         if not hasattr(obj, filter_key) or \
91                 filter_key in ['hash', 'updatetime', 'createtime', 'events']:
92             raise BadRequestException(
93                 'Filter attrName {} is invalid'.format(filter_key))