我想构建一个简单的“DataSource”类,其属性名为“data_source_type”和“data_source_path”。其中“data_source_type”是一个枚举,“data_source_path”是一个字符串,取决于“data_source_type”,我想将“ValidFilePath”或“ValidHttpURL”等适当的验证设置为“data_source_path”。
我不想编写 IF-ELSE 并有一个意大利面条数据源类,我想利用“Python 描述符”或任何其他优雅的 Python 构造,它们会考虑 SRP(单一责任原则)并支持函数式编程构造。
数据源.py
1 import re
2 from enum import Enum
3 import os
4
5
6 class ValidFilePath(object):
7 def __set__(self, obj, val):
8 if not os.path.exists():
9 raise ValueError("Please enter a valid file path")
10 self.__url = val
11
12 def __get__(self, obj, objtype):
13 return self.__url
14
15
16 class ValidHttpURL(object):
17 def __set__(self, obj, val):
18 if (val is None or re.compile(
19 r'^https?://' # http:// or https://
20 r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' # domain...
21 r'localhost|' # localhost...
22 r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip
23 r'(?::\d+)?' # optional port
24 r'(?:/?|[/?]\S+)$', re.IGNORECASE).search(val) is None):
25 raise ValueError("Please set an valid HTTP(S) URL")
26 self.__url = val
27
28 def __get__(self, obj, objtype):
29 return self.__url
30
31
32 class DataSourceType(Enum):
33 HTTP = 100,
34 LOCAL_FILE = 200,
35 HDFS_FILE = 300
36
37
38 class ValidDataSourceType(object):
39 def __set__(self, obj, val):
40 if val is None or not DataSourceType.__contains__(DataSourceType[val]):
41 raise ValueError("Please set a valid Data Source Type Enum, "
42 " possible values are -> ", [e.name for e in DataSourceType])
43 self.__data_source_type = DataSourceType[val]
44
45 def __get__(self, obj, objtype):
46 return self.__data_source_type
47
48
49 class DataSource(object):
50 data_source_type = ValidDataSourceType()
51 data_source_path = ValidHttpURL()
截至目前,在第 51 行中,我放置了“ValidHttpURL”,我想根据“data_source_type”设置适当的验证描述符
预期行为
ds1 = DataSource()
ds1.data_source_type = 'HTTP'
ds1.data_source_path = 'http://www.google.com'
ds2 = DataSource()
ds2.data_source_type = 'LOCAL_FILE'
ds2.data_source_path = '/var/www/index.html'
print("All is well")
实际行为
ds1 = DataSource()
ds1.data_source_type = 'HTTP'
ds1.data_source_path = 'http://www.google.com'
ds2 = DataSource()
ds2.data_source_type = 'LOCAL_FILE'
ds2.data_source_path = '/var/www/index.html'
**ValueError: Please set an valid HTTP(S) URL**
***更新的答案***
1 import os
2 import re
3 from enum import Enum
4 from weakref import WeakKeyDictionary
5
6
7 def valid_file_path(value):
8 if not os.path.exists(value):
9 raise ValueError(value, " is not present. Please make sure the file exists")
10 return value
11
12
13 def valid_http_url(value):
14 if (value is None or re.compile(
15 r'^https?://' # http:// or https://
16 r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' # domain...
17 r'localhost|' # localhost...
18 r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip
19 r'(?::\d+)?' # optional port
20 r'(?:/?|[/?]\S+)$', re.IGNORECASE).search(value) is None):
21 raise ValueError("Please set an valid HTTP(S) URL")
22 return value
23
24
25 class DataSourceType(Enum):
26 NOT_DEFINED = (0, None)
27 HTTP = (100, valid_http_url)
28 LOCAL_FILE = (200, valid_file_path)
29
30 def __init__(self, enum_id, enum_validator):
31 self._id = enum_id
32 self._validator = enum_validator
33
34 @property
35 def validator(self):
36 return self._validator
37
38
39 class ValidDataSourceType(object):
40 def __init__(self):
41 self.default = DataSourceType.NOT_DEFINED
42 self.values = WeakKeyDictionary()
43
44 def __get__(self, instance, owner):
45 return self.values.get(instance, self.default)
46
47 def __set__(self, instance, value):
48 if value is None or not DataSourceType.__contains__(DataSourceType[value]):
49 raise ValueError("Please set a valid Data Source Type Enum, "
50 " possible values are -> ", [e.name for e in DataSourceType])
51 self.values[instance] = DataSourceType[value]
52
53 def __delete__(self, instance):
54 del self.values[instance]
55
56
57 class ValidDataSourcePath(object):
58 def __init__(self, default_data_source_type_field='data_source_type'):
59 self._default = ''
60 self._default_data_source_type_field = default_data_source_type_field
61 self.values = WeakKeyDictionary()
62
63 def __get__(self, instance, owner):
64 return self.values.get(instance, self._default)
65
66 def __set__(self, instance, *value):
67 data_source_type_field = self._default_data_source_type_field
68 value_to_set = None
69
70 if value and len(value) == 1 and isinstance(value[0], str): # user sent only the value
71 value_to_set = value[0]
72 if value and len(value) == 1 and isinstance(value[0], tuple): # user sent the value , and the validation field
73 value_to_set = value[0][0]
74 data_source_type_field = value[0][1]
75
76 _data_source_type = getattr(instance, data_source_type_field, None)
77 if not _data_source_type:
78 raise ValueError(" Valid source path depends on ValidDataSourceType , "
79 " please make sure you have an attribute named ValidDataSourceType")
80 _data_source_type.validator(value_to_set)
81 self.values[instance] = value_to_set
82
83
84 class DataSource(object):
85 data_source_type = ValidDataSourceType()
86 data_source_path = ValidDataSourcePath()
87
88
89 class SomeOtherDomainModel(object):
90 data_source_type_ext = ValidDataSourceType()
91 data_source_path = ValidDataSourcePath()
92
93
94 print(" **************** Scenario 1 - Start **************** ")
95 ds1 = DataSource()
96 ds1.data_source_type = 'HTTP'
97 ds1.data_source_path = "http://www.google.com"
98 print(ds1.data_source_path)
99 print(" **************** Scenario 1 - End **************** ")
100
101 print(" **************** Scenario 2 - Start **************** ")
102 ds2 = SomeOtherDomainModel()
103 ds2.data_source_type_ext = 'HTTP'
104 ds2.data_source_path = ("http://www.yahoo.com", 'data_source_type_ext')
105 print(ds2.data_source_path)
106 print(" **************** Scenario 2 - Start **************** ")