0
# Error Handling
1

2
PyKwalify provides a structured exception hierarchy with specific error types for different validation failures, providing detailed error messages and validation paths for debugging and error reporting.
3

4
## Capabilities
5

6
### Base Exception Class
7

8
```python { .api }
9
class PyKwalifyException(RuntimeError):
10
    """Base exception class for all PyKwalify errors."""
11
    
12
    def __init__(self, msg=None, error_key=None, retcode=None, path=None):
13
        """
14
        Initialize PyKwalify exception.
15
        
16
        Args:
17
            msg (str, optional): Error message
18
            error_key (str, optional): Unique error identifier
19
            retcode (int, optional): Error return code
20
            path (str, optional): Path where error occurred (default: "/")
21
        """
22
    
23
    @property
24
    def msg(self):
25
        """Error message string"""
26
    
27
    @property
28
    def retcode(self):
29
        """Error return code integer"""
30
    
31
    @property
32
    def retname(self):
33
        """Error return code name string"""
34
```
35

36
### Validation Error Classes
37

38
```python { .api }
39
class SchemaError(PyKwalifyException):
40
    """Raised when schema validation fails."""
41
    
42
    class SchemaErrorEntry(object):
43
        """Individual schema error entry."""
44
        def __init__(self, msg, path, value, **kwargs):
45
            """
46
            Args:
47
                msg (str): Error message template
48
                path (str): Path where error occurred
49
                value: Value that caused the error
50
                **kwargs: Additional error context
51
            """
52

53
class CoreError(PyKwalifyException):
54
    """Raised when core processing encounters an error."""
55

56
class RuleError(PyKwalifyException):
57
    """Raised when rule processing encounters an error."""
58

59
class NotMappingError(PyKwalifyException):
60
    """Raised when a value is not a mapping when expected."""
61

62
class NotSequenceError(PyKwalifyException):
63
    """Raised when a value is not a sequence when expected."""
64

65
class SchemaConflict(PyKwalifyException):
66
    """Raised when schema definitions conflict."""
67

68
class UnknownError(PyKwalifyException):
69
    """Raised for unknown or unexpected errors."""
70
```
71

72
### Error Code Constants
73

74
```python { .api }
75
retcodes: dict  # Mapping of error codes to names
76
"""
77
{
78
    0: 'noerror',
79
    1: 'unknownerror',
80
    2: 'schemaerror',
81
    3: 'coreerror',
82
    4: 'ruleerror',
83
    5: 'schemaconflict',
84
    6: 'notmaperror',
85
    7: 'notsequenceerror',
86
}
87
"""
88

89
retnames: dict  # Mapping of error names to codes
90
"""
91
{
92
    'noerror': 0,
93
    'unknownerror': 1,
94
    'schemaerror': 2,
95
    'coreerror': 3,
96
    'ruleerror': 4,
97
    'schemaconflict': 5,
98
    'notmaperror': 6,
99
    'notsequenceerror': 7,
100
}
101
"""
102
```
103

104
## Usage Examples
105

106
### Basic Error Handling
107

108
```python
109
from pykwalify.core import Core
110
from pykwalify.errors import SchemaError, CoreError
111

112
# Invalid data that will fail validation
113
data = {
114
    "name": 123,  # Should be string
115
    "age": "thirty"  # Should be integer
116
}
117

118
schema = {
119
    "type": "map",
120
    "mapping": {
121
        "name": {"type": "str", "required": True},
122
        "age": {"type": "int", "required": True}
123
    }
124
}
125

126
try:
127
    c = Core(source_data=data, schema_data=schema)
128
    c.validate(raise_exception=True)
129
    
130
except SchemaError as e:
131
    print(f"Schema validation failed!")
132
    print(f"Error message: {e.msg}")
133
    print(f"Error path: {e.path}")
134
    print(f"Error code: {e.retcode}")
135
    print(f"Error name: {e.retname}")
136
    
137
except CoreError as e:
138
    print(f"Core processing error: {e.msg}")
139
```
140

141
### Comprehensive Error Handling
142

143
```python
144
from pykwalify.core import Core
145
from pykwalify.errors import (
146
    PyKwalifyException, SchemaError, CoreError, RuleError,
147
    NotMappingError, NotSequenceError, SchemaConflict, UnknownError
148
)
149

150
def validate_with_comprehensive_error_handling(data, schema):
151
    try:
152
        c = Core(source_data=data, schema_data=schema)
153
        c.validate(raise_exception=True)
154
        return True, "Validation successful"
155
        
156
    except SchemaError as e:
157
        return False, f"Schema validation failed: {e.msg} (Path: {e.path})"
158
        
159
    except NotMappingError as e:
160
        return False, f"Expected mapping but got different type: {e.msg} (Path: {e.path})"
161
        
162
    except NotSequenceError as e:
163
        return False, f"Expected sequence but got different type: {e.msg} (Path: {e.path})"
164
        
165
    except RuleError as e:
166
        return False, f"Rule processing error: {e.msg} (Path: {e.path})"
167
        
168
    except SchemaConflict as e:
169
        return False, f"Schema conflict: {e.msg} (Path: {e.path})"
170
        
171
    except CoreError as e:
172
        return False, f"Core processing error: {e.msg}"
173
        
174
    except UnknownError as e:
175
        return False, f"Unknown error: {e.msg}"
176
        
177
    except PyKwalifyException as e:
178
        return False, f"PyKwalify error: {e.msg} (Code: {e.retcode})"
179
        
180
    except Exception as e:
181
        return False, f"Unexpected error: {str(e)}"
182

183
# Test with various error scenarios
184
test_cases = [
185
    # Valid data
186
    ({"name": "John", "age": 30}, {"type": "map", "mapping": {"name": {"type": "str"}, "age": {"type": "int"}}}),
187
    
188
    # Schema error - wrong type
189
    ({"name": 123}, {"type": "map", "mapping": {"name": {"type": "str"}}}),
190
    
191
    # Not mapping error - expecting dict but got list
192
    ([1, 2, 3], {"type": "map", "mapping": {"key": {"type": "str"}}}),
193
    
194
    # Not sequence error - expecting list but got dict
195
    ({"key": "value"}, {"type": "seq", "sequence": [{"type": "str"}]}),
196
]
197

198
for i, (test_data, test_schema) in enumerate(test_cases):
199
    success, message = validate_with_comprehensive_error_handling(test_data, test_schema)
200
    print(f"Test {i+1}: {'PASS' if success else 'FAIL'} - {message}")
201
```
202

203
### Accessing Detailed Error Information
204

205
```python
206
from pykwalify.core import Core
207
from pykwalify.errors import SchemaError
208

209
# Complex data with multiple validation errors
210
data = {
211
    "user": {
212
        "name": "",  # Too short
213
        "age": -5,   # Below minimum
214
        "email": "invalid-email"  # Invalid format
215
    },
216
    "items": [
217
        {"id": "abc", "count": "five"},  # id should be int, count should be int
218
        {"id": 123}  # missing required count
219
    ]
220
}
221

222
schema = {
223
    "type": "map",
224
    "mapping": {
225
        "user": {
226
            "type": "map",
227
            "mapping": {
228
                "name": {"type": "str", "length": {"min": 1}},
229
                "age": {"type": "int", "range": {"min": 0}},
230
                "email": {"type": "email"}
231
            }
232
        },
233
        "items": {
234
            "type": "seq",
235
            "sequence": [{
236
                "type": "map",
237
                "mapping": {
238
                    "id": {"type": "int", "required": True},
239
                    "count": {"type": "int", "required": True}
240
                }
241
            }]
242
        }
243
    }
244
}
245

246
c = Core(source_data=data, schema_data=schema)
247

248
# Validate without raising exceptions to collect all errors
249
is_valid = c.validate(raise_exception=False)
250

251
if not is_valid:
252
    print("Validation failed with the following errors:")
253
    
254
    # Access general error list
255
    if c.errors:
256
        print("\nGeneral errors:")
257
        for error in c.errors:
258
            print(f"  - {error}")
259
    
260
    # Access validation errors
261
    if c.validation_errors:
262
        print("\nValidation errors:")
263
        for error in c.validation_errors:
264
            print(f"  - {error}")
265
    
266
    # Access validation error exceptions for detailed info
267
    if c.validation_errors_exceptions:
268
        print("\nDetailed validation errors:")
269
        for exc in c.validation_errors_exceptions:
270
            print(f"  - {exc.msg}")
271
            print(f"    Path: {exc.path}")
272
            print(f"    Code: {exc.retcode} ({exc.retname})")
273
            if hasattr(exc, 'error_key') and exc.error_key:
274
                print(f"    Key: {exc.error_key}")
275
```
276

277
### File-Based Error Handling
278

279
```python
280
from pykwalify.core import Core
281
from pykwalify.errors import CoreError, SchemaError
282
import tempfile
283
import os
284

285
def validate_files_with_error_handling(data_content, schema_content):
286
    # Create temporary files
287
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as data_file:
288
        data_file.write(data_content)
289
        data_file_path = data_file.name
290
    
291
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as schema_file:
292
        schema_file.write(schema_content)
293
        schema_file_path = schema_file.name
294
    
295
    try:
296
        c = Core(source_file=data_file_path, schema_files=[schema_file_path])
297
        c.validate(raise_exception=True)
298
        return True, "File validation successful"
299
        
300
    except CoreError as e:
301
        if "do not exists on disk" in str(e.msg):
302
            return False, f"File not found: {e.msg}"
303
        elif "Unable to load" in str(e.msg):
304
            return False, f"File loading error: {e.msg}"
305
        else:
306
            return False, f"Core error: {e.msg}"
307
            
308
    except SchemaError as e:
309
        return False, f"Schema validation error: {e.msg} at {e.path}"
310
        
311
    finally:
312
        # Clean up temporary files
313
        try:
314
            os.unlink(data_file_path)
315
            os.unlink(schema_file_path)
316
        except:
317
            pass
318

319
# Test file validation
320
data_yaml = """
321
name: John Doe
322
age: 30
323
email: john@example.com
324
"""
325

326
schema_yaml = """
327
type: map
328
mapping:
329
  name: {type: str, required: true}
330
  age: {type: int, range: {min: 0, max: 120}}
331
  email: {type: email}
332
"""
333

334
success, message = validate_files_with_error_handling(data_yaml, schema_yaml)
335
print(f"File validation: {'SUCCESS' if success else 'FAILED'} - {message}")
336
```
337

338
### Custom Error Handling with Context
339

340
```python
341
from pykwalify.core import Core
342
from pykwalify.errors import SchemaError, PyKwalifyException
343

344
class ValidationResult:
345
    def __init__(self, success=True, errors=None, warnings=None):
346
        self.success = success
347
        self.errors = errors or []
348
        self.warnings = warnings or []
349
    
350
    def add_error(self, error_type, message, path=None):
351
        self.errors.append({
352
            'type': error_type,
353
            'message': message,
354
            'path': path
355
        })
356
        self.success = False
357
    
358
    def __str__(self):
359
        if self.success:
360
            return "Validation successful"
361
        
362
        result = f"Validation failed with {len(self.errors)} error(s):"
363
        for error in self.errors:
364
            path_info = f" at {error['path']}" if error['path'] else ""
365
            result += f"\n  - {error['type']}: {error['message']}{path_info}"
366
        return result
367

368
def advanced_validate(data, schema, context=None):
369
    """Enhanced validation with custom error handling and context."""
370
    result = ValidationResult()
371
    context = context or {}
372
    
373
    try:
374
        c = Core(source_data=data, schema_data=schema)
375
        
376
        # Add context to error messages if validation fails
377
        try:
378
            c.validate(raise_exception=True)
379
        except PyKwalifyException as e:
380
            error_context = context.get('error_context', 'validation')
381
            result.add_error(
382
                error_type=e.__class__.__name__,
383
                message=f"[{error_context}] {e.msg}",
384
                path=getattr(e, 'path', None)
385
            )
386
            
387
            # Add suggestions based on error type
388
            if isinstance(e, SchemaError):
389
                if 'type' in str(e.msg).lower():
390
                    result.add_error(
391
                        'Suggestion',
392
                        'Check that data types match schema requirements',
393
                        path=getattr(e, 'path', None)
394
                    )
395
    
396
    except Exception as e:
397
        result.add_error('UnexpectedError', str(e))
398
    
399
    return result
400

401
# Test advanced validation
402
test_data = {"name": 123, "age": "thirty"}
403
test_schema = {
404
    "type": "map",
405
    "mapping": {
406
        "name": {"type": "str"},
407
        "age": {"type": "int"}
408
    }
409
}
410

411
validation_result = advanced_validate(
412
    test_data, 
413
    test_schema, 
414
    context={'error_context': 'user_profile_validation'}
415
)
416

417
print(validation_result)
418
```
419

420
## Error Recovery Patterns
421

422
### Validation with Fallbacks
423

424
```python
425
from pykwalify.core import Core
426
from pykwalify.errors import SchemaError
427

428
def validate_with_fallback(data, primary_schema, fallback_schema=None):
429
    """Try primary schema, fall back to secondary if validation fails."""
430
    
431
    # Try primary schema
432
    try:
433
        c = Core(source_data=data, schema_data=primary_schema)
434
        c.validate(raise_exception=True)
435
        return True, "Primary schema validation successful", None
436
        
437
    except SchemaError as primary_error:
438
        if fallback_schema:
439
            try:
440
                c = Core(source_data=data, schema_data=fallback_schema)
441
                c.validate(raise_exception=True)
442
                return True, "Fallback schema validation successful", primary_error
443
                
444
            except SchemaError as fallback_error:
445
                return False, "Both schemas failed validation", [primary_error, fallback_error]
446
        else:
447
            return False, "Primary schema validation failed", primary_error
448

449
# Example usage
450
data = {"version": "1.0", "name": "test"}
451

452
# Strict schema (v2 format)
453
strict_schema = {
454
    "type": "map",
455
    "mapping": {
456
        "version": {"type": "str", "enum": ["2.0"]},
457
        "name": {"type": "str"},
458
        "description": {"type": "str", "required": True}
459
    }
460
}
461

462
# Lenient schema (v1 format)
463
lenient_schema = {
464
    "type": "map",
465
    "mapping": {
466
        "version": {"type": "str"},
467
        "name": {"type": "str", "required": True}
468
    }
469
}
470

471
success, message, errors = validate_with_fallback(data, strict_schema, lenient_schema)
472
print(f"Result: {message}")
473
if errors:
474
    print("Original error:", errors.msg if hasattr(errors, 'msg') else str(errors))
475
```