Coverage for src/dataknobs_data/validation_v2/factory.py: 68%

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

2 

3import logging 

4from typing import Any, Dict, List, Optional 

5 

6from dataknobs_config import FactoryBase 

7 

8from .schema import Schema 

9from .constraints import ( 

10 Constraint, 

11 Required, 

12 Range, 

13 Length, 

14 Pattern, 

15 Enum, 

16 Unique, 

17 Custom, 

18 All, 

19 Any, 

20) 

21from .coercer import Coercer 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26class SchemaFactory(FactoryBase): 

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

28  

29 Configuration Options: 

30 name (str): Schema name 

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

32 description (str): Optional schema description 

33 fields (list): List of field definitions 

34  

35 Field Definition Options: 

36 name (str): Field name 

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

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

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

40 description (str): Field description 

41 constraints (list): List of constraint definitions 

42  

43 Example Configuration: 

44 schemas: 

45 - name: user_schema 

46 factory: schema 

47 strict: true 

48 description: User registration schema 

49 fields: 

50 - name: username 

51 type: STRING 

52 required: true 

53 constraints: 

54 - type: length 

55 min: 3 

56 max: 20 

57 - type: pattern 

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

59 - name: age 

60 type: INTEGER 

61 constraints: 

62 - type: range 

63 min: 13 

64 max: 120 

65 """ 

66 

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

68 """Create a Schema instance from configuration. 

69  

70 Args: 

71 **config: Schema configuration 

72  

73 Returns: 

74 Schema instance 

75 """ 

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

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

78 description = config.get("description") 

79 

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

81 

82 schema = Schema(name, strict) 

83 if description: 

84 schema.with_description(description) 

85 

86 # Add fields 

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

88 for field_config in fields: 

89 self._add_field_to_schema(schema, field_config) 

90 

91 return schema 

92 

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

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

95  

96 Args: 

97 schema: Schema to add field to 

98 field_config: Field configuration 

99 """ 

100 from dataknobs_data.fields import FieldType 

101 

102 field_name = field_config.get("name") 

103 if not field_name: 

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

105 return 

106 

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

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

109 default = field_config.get("default") 

110 description = field_config.get("description") 

111 

112 # Build constraints 

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

114 

115 schema.field( 

116 name=field_name, 

117 field_type=field_type, 

118 required=required, 

119 default=default, 

120 constraints=constraints, 

121 description=description 

122 ) 

123 

124 def _build_constraints(self, constraint_configs: List[Dict[str, Any]]) -> List[Constraint]: 

125 """Build constraint objects from configuration. 

126  

127 Args: 

128 constraint_configs: List of constraint configurations 

129  

130 Returns: 

131 List of Constraint objects 

132 """ 

133 constraints = [] 

134 

135 for config in constraint_configs: 

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

137 

138 if constraint_type == "required": 

139 constraints.append(Required( 

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

141 )) 

142 

143 elif constraint_type == "range": 

144 constraints.append(Range( 

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

146 max=config.get("max") 

147 )) 

148 

149 elif constraint_type == "length": 

150 constraints.append(Length( 

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

152 max=config.get("max") 

153 )) 

154 

155 elif constraint_type == "pattern": 

156 pattern = config.get("pattern") 

157 if pattern: 

158 constraints.append(Pattern(pattern)) 

159 

160 elif constraint_type == "enum": 

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

162 if values: 

163 constraints.append(Enum(values)) 

164 

165 elif constraint_type == "unique": 

166 constraints.append(Unique( 

167 field_name=config.get("field_name") 

168 )) 

169 

170 elif constraint_type == "all": 

171 # Recursive build for composite constraints 

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

173 if sub_constraints: 

174 constraints.append(All(sub_constraints)) 

175 

176 elif constraint_type == "any": 

177 # Recursive build for composite constraints 

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

179 if sub_constraints: 

180 constraints.append(Any(sub_constraints)) 

181 

182 else: 

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

184 

185 return constraints 

186 

187 

188class CoercerFactory(FactoryBase): 

189 """Factory for creating Coercer instances. 

190  

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

192 provides a consistent interface for the config system. 

193 """ 

194 

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

196 """Create a Coercer instance. 

197  

198 Args: 

199 **config: Currently unused 

200  

201 Returns: 

202 Coercer instance 

203 """ 

204 logger.info("Creating Coercer") 

205 return Coercer() 

206 

207 

208# Create singleton instances for registration 

209schema_factory = SchemaFactory() 

210coercer_factory = CoercerFactory()