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
  21 from o2common.helper import o2logging
  22 logger = o2logging.get_logger(__name__)
  23
  24
  25 def gen_filter(obj: ColumnElement, filter_str: str):
  26     check_filter(obj, filter_str)
  27     try:
  28         filter_list = gen_orm_filter(obj, filter_str)
  29     except KeyError as e:
  30         raise BadRequestException(e.args[0])
  31     return filter_list
  32
  33
  34 # The regular expressions testing example put on here
  35 # (neq,testkey,value-1)
  36 # (neq,testkey,value-1,value-2)
  37 # (gt,hello,1)
  38 # (gte,world,2)
  39 # (lt,testlt,notint)
  40 # (ncont,key1,v1,v_2)
  41 # (gt,hello,1);(ncont,world,val1,val-2)
  42 # (eq,wrong,60cba7be-e2cd-3b8c-a7ff-16e0f10573f9)
  43 # (eq,description,value key)
  44 def check_filter(obj: ColumnElement, filter_str: str):
  45     if not filter_str:
  46         return
  47     # pattern = r'^(\((eq|neq|gt|lt|gte|lte){1},\w+,[\w -\.]+\)\;?|' +\
  48     #     r'\((in|nin|cont|ncont){1},\w*(,[\w -\.]*)*\)\;?)+'
  49     # result = re.match(pattern, filter_str)
  50     # logger.debug('filter: {} match result is {}'.format(filter_str, result))
  51     # if not result:
  52     #     raise BadRequestException(
  53     #         'filter value format is invalid')
  54     check_filter_attribute(obj, filter_str)
  55
  56
  57 def check_filter_attribute(obj: ColumnElement, filter_str: str):
  58     # filter_without_space = filter_str.replace(" ", "")
  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         # if '(' in i:
  66         #     i = i.replace("(", "")
  67         # if ')' in i:
  68         #     i = i.replace(")", "")
  69         filter_expr = i.split(',')
  70         if len(filter_expr) < 3:
  71             raise BadRequestException(
  72                 'ignore invalid filter {}'.format(i))
  73             continue
  74         filter_op = filter_expr[0].strip()
  75         filter_key = filter_expr[1].strip()
  76         filter_vals = filter_expr[2:]
  77         if filter_op in ["eq", "neq", "gt", "lt", "gte", "lte"]:
  78             if len(filter_vals) != 1:
  79                 raise BadRequestException(
  80                     "Found {} values: {} while only single value"
  81                     " is allowed for operation {}".format(
  82                         len(filter_vals), filter_vals, filter_op)
  83                 )
  84         elif filter_op not in ["in", "nin", "cont", "ncont"]:
  85             raise BadRequestException(
  86                 'Filter operation {} is invalid'.format(filter_op)
  87             )
  88         else:
  89             pass
  90         if not hasattr(obj, filter_key):
  91             raise BadRequestException(
  92                 'Filter attrName {} is invalid'.format(filter_key))