Coverage for src / dataknobs_data / validation / factory.py: 22%

1"""Factory classes for validation v2 components.""" 

2 

3import logging 

4from typing import Any 

5 

6from dataknobs_config import FactoryBase 

7 

8from .coercer import Coercer 

9from .constraints import ( 

10 All, 

11 AnyOf, 

12 Constraint, 

13 Enum, 

14 Length, 

15 Pattern, 

16 Range, 

17 Required, 

18 Unique, 

19) 

20from .schema import Schema 

21 

22logger = logging.getLogger(__name__) 

23 

24 

25class SchemaFactory(FactoryBase): 

26 """Factory for creating validation schemas from configuration. 

27  

28 Configuration Options: 

29 name (str): Schema name 

30 strict (bool): Whether to reject unknown fields (default: False) 

31 description (str): Optional schema description 

32 fields (list): List of field definitions 

33  

34 Field Definition Options: 

35 name (str): Field name 

36 type (str): Field type (STRING, INTEGER, FLOAT, BOOLEAN, DATETIME, JSON, BINARY) 

37 required (bool): Whether field is required (default: False) 

38 default (any): Default value if field is missing 

39 description (str): Field description 

40 constraints (list): List of constraint definitions 

41  

42 Example Configuration: 

43 schemas: 

44 - name: user_schema 

45 factory: schema 

46 strict: true 

47 description: User registration schema 

48 fields: 

49 - name: username 

50 type: STRING 

51 required: true 

52 constraints: 

53 - type: length 

54 min: 3 

55 max: 20 

56 - type: pattern 

57 pattern: "^[a-zA-Z0-9_]+$" 

58 - name: age 

59 type: INTEGER 

60 constraints: 

61 - type: range 

62 min: 13 

63 max: 120 

64 """ 

65 

66 def create(self, **config) -> Schema: 

67 """Create a Schema instance from configuration. 

68  

69 Args: 

70 **config: Schema configuration 

71  

72 Returns: 

73 Schema instance 

74 """ 

75 name = config.get("name", "unnamed_schema") 

76 strict = config.get("strict", False) 

77 description = config.get("description") 

78 

79 logger.info(f"Creating schema: {name}") 

80 

81 schema = Schema(name, strict) 

82 if description: 

83 schema.with_description(description) 

84 

85 # Add fields 

86 fields = config.get("fields", []) 

87 for field_config in fields: 

88 self._add_field_to_schema(schema, field_config) 

89 

90 return schema 

91 

92 def _add_field_to_schema(self, schema: Schema, field_config: dict[str, Any]) -> None: 

93 """Add a field to the schema based on configuration. 

94  

95 Args: 

96 schema: Schema to add field to 

97 field_config: Field configuration 

98 """ 

99 field_name = field_config.get("name") 

100 if not field_name: 

101 logger.warning("Field configuration missing 'name', skipping") 

102 return 

103 

104 field_type = field_config.get("type", "STRING") 

105 required = field_config.get("required", False) 

106 default = field_config.get("default") 

107 description = field_config.get("description") 

108 

109 # Build constraints 

110 constraints = self._build_constraints(field_config.get("constraints", [])) 

111 

112 schema.field( 

113 name=field_name, 

114 field_type=field_type, 

115 required=required, 

116 default=default, 

117 constraints=constraints, 

118 description=description 

119 ) 

120 

121 def _build_constraints(self, constraint_configs: list[dict[str, Any]]) -> list[Constraint]: 

122 """Build constraint objects from configuration. 

123  

124 Args: 

125 constraint_configs: List of constraint configurations 

126  

127 Returns: 

128 List of Constraint objects 

129 """ 

130 constraints: list[Constraint] = [] 

131 

132 for config in constraint_configs: 

133 constraint_type = config.get("type", "").lower() 

134 

135 if constraint_type == "required": 

136 constraints.append(Required( 

137 allow_empty=config.get("allow_empty", False) 

138 )) 

139 

140 elif constraint_type == "range": 

141 constraints.append(Range( 

142 min=config.get("min"), 

143 max=config.get("max") 

144 )) 

145 

146 elif constraint_type == "length": 

147 constraints.append(Length( 

148 min=config.get("min"), 

149 max=config.get("max") 

150 )) 

151 

152 elif constraint_type == "pattern": 

153 pattern = config.get("pattern") 

154 if pattern: 

155 constraints.append(Pattern(pattern)) 

156 

157 elif constraint_type == "enum": 

158 values = config.get("values", []) 

159 if values: 

160 constraints.append(Enum(values)) 

161 

162 elif constraint_type == "unique": 

163 constraints.append(Unique( 

164 field_name=config.get("field_name") 

165 )) 

166 

167 elif constraint_type == "all": 

168 # Recursive build for composite constraints 

169 sub_constraints = self._build_constraints(config.get("constraints", [])) 

170 if sub_constraints: 

171 constraints.append(All(sub_constraints)) 

172 

173 elif constraint_type == "any": 

174 # Recursive build for composite constraints 

175 sub_constraints = self._build_constraints(config.get("constraints", [])) 

176 if sub_constraints: 

177 constraints.append(AnyOf(sub_constraints)) 

178 

179 else: 

180 logger.warning(f"Unknown constraint type: {constraint_type}") 

181 

182 return constraints 

183 

184 

185class CoercerFactory(FactoryBase): 

186 """Factory for creating Coercer instances. 

187  

188 The Coercer doesn't require configuration, but this factory 

189 provides a consistent interface for the config system. 

190 """ 

191 

192 def create(self, **config) -> Coercer: 

193 """Create a Coercer instance. 

194  

195 Args: 

196 **config: Currently unused 

197  

198 Returns: 

199 Coercer instance 

200 """ 

201 logger.info("Creating Coercer") 

202 return Coercer() 

203 

204 

205# Create singleton instances for registration 

206schema_factory = SchemaFactory() 

207coercer_factory = CoercerFactory()