1# Copyright (C) 2020 Bloomberg LP 

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 

16from abc import ABC 

17from datetime import datetime 

18import re 

19 

20import dateutil.parser 

21from dateutil.tz import tzutc 

22 

23import buildgrid._enums as enums 

24from buildgrid._exceptions import InvalidArgumentError 

25from buildgrid.server.operations.filtering import SortKey 

26 

27 

28class ValueSanitizer(ABC): 

29 """ Base sanitizer class. """ 

30 def sanitize(self, value_string: str): 

31 """ Method that takes an input string, validates that input string, 

32 and transforms it to a value of another type if necessary. 

33 

34 Raises InvalidArgumentError if the sanitization fails. Returns a 

35 value of an arbitrary type if it succeeds. """ 

36 return NotImplementedError() 

37 

38 

39class RegexValueSanitizer(ValueSanitizer): 

40 """ Sanitizer for regexable patterns. """ 

41 def __init__(self, filter_name, regex_pattern): 

42 self.filter_name = filter_name 

43 self.regex = re.compile(f"^({regex_pattern})$") 

44 

45 def sanitize(self, value_string: str) -> str: 

46 if not self.regex.fullmatch(value_string): 

47 raise InvalidArgumentError(f"[{value_string}] is not a valid value for {self.filter_name}.") 

48 return value_string 

49 

50 

51class DatetimeValueSanitizer(ValueSanitizer): 

52 """ Sanitizer for ISO 8601 datetime strings. """ 

53 def __init__(self, filter_name): 

54 self.filter_name = filter_name 

55 

56 def sanitize(self, value_string: str) -> datetime: 

57 try: 

58 dt = dateutil.parser.isoparse(value_string) 

59 # Convert to UTC and remove timezone awareness 

60 if dt.tzinfo: 

61 dt = dt.astimezone(tz=tzutc()).replace(tzinfo=None) 

62 return dt 

63 except ValueError: 

64 raise InvalidArgumentError(f"[{value_string}] is not a valid value for {self.filter_name}.") 

65 

66 

67class OperationStageValueSanitizer(ValueSanitizer): 

68 """ Sanitizer for the OperationStage type. 

69 

70 Matches valid OperationStage values and converts to the 

71 numeric representation of that stage. """ 

72 def __init__(self, filter_name): 

73 self.filter_name = filter_name 

74 

75 def sanitize(self, value_string: str) -> int: 

76 try: 

77 stage = value_string.upper() 

78 return enums.OperationStage[stage].value 

79 except KeyError: 

80 raise InvalidArgumentError(f"[{value_string}] is not a valid value for {self.filter_name}.") 

81 

82 

83class SortKeyValueSanitizer(ValueSanitizer): 

84 """ Sanitizer for sort orders. 

85 

86 Produces a SortKey tuple, which specifies both a key name and a boolean 

87 indicating ascending/descending order. """ 

88 def __init__(self, filter_name): 

89 self.filter_name = filter_name 

90 

91 def sanitize(self, value_string: str) -> SortKey: 

92 desc_key = "(desc)" 

93 asc_key = "(asc)" 

94 

95 key_name = value_string.lower().strip() 

96 if not key_name: 

97 raise InvalidArgumentError(f"Invalid sort key [{value_string}].") 

98 descending = False 

99 

100 if value_string.endswith(desc_key): 

101 descending = True 

102 key_name = key_name[:-len(desc_key)].strip() 

103 if not key_name: 

104 raise InvalidArgumentError(f"Invalid sort key [{value_string}].") 

105 

106 elif value_string.endswith(asc_key): 

107 key_name = key_name[:-len(asc_key)].strip() 

108 if not key_name: 

109 raise InvalidArgumentError(f"Invalid sort key [{value_string}].") 

110 

111 return SortKey(name=key_name, descending=descending)