Coverage for queue_cli/core/schema.py: 0.00%

1""" 

2Schema validation and translation for match messages. 

3 

4Handles validation against the canonical schema and translation 

5between the external schema (match-scraper) and internal schema (celery tasks). 

6""" 

7 

8import json 

9from pathlib import Path 

10from typing import Any 

11 

12import jsonschema 

13from jsonschema import ValidationError 

14 

15 

16class SchemaValidator: 

17 """Validates and translates match message schemas.""" 

18 

19 # Field name mappings from canonical schema to internal schema 

20 FIELD_MAPPINGS = { 

21 "date": "match_date", 

22 "score_home": "home_score", 

23 "score_away": "away_score", 

24 "status": "match_status", 

25 "match_id": "external_match_id", 

26 } 

27 

28 def __init__(self, schema_path: str | None = None): 

29 """ 

30 Initialize schema validator. 

31 

32 Args: 

33 schema_path: Path to JSON schema file. If None, uses default schema. 

34 """ 

35 if schema_path is None: 

36 # Default to canonical schema in docs 

37 schema_path = str( 

38 Path(__file__).parent.parent.parent.parent / "docs" / "08-integrations" / "match-message-schema.json" 

39 ) 

40 

41 with open(schema_path) as f: 

42 self.schema = json.load(f) 

43 

44 def validate(self, data: dict[str, Any]) -> tuple[bool, list[str], list[str]]: 

45 """ 

46 Validate data against schema. 

47 

48 Args: 

49 data: Data to validate 

50 

51 Returns: 

52 Tuple of (is_valid, errors, warnings) 

53 """ 

54 errors = [] 

55 warnings = [] 

56 

57 try: 

58 jsonschema.validate(instance=data, schema=self.schema) 

59 except ValidationError as e: 

60 errors.append(f"Schema validation failed: {e.message}") 

61 return False, errors, warnings 

62 

63 # Check for fields that need translation 

64 for canonical_field, internal_field in self.FIELD_MAPPINGS.items(): 

65 if canonical_field in data and internal_field in data: 

66 warnings.append(f"Both '{canonical_field}' and '{internal_field}' present. Using '{internal_field}'.") 

67 

68 return True, errors, warnings 

69 

70 def translate_to_internal(self, data: dict[str, Any]) -> dict[str, Any]: 

71 """ 

72 Translate from canonical schema (match-scraper) to internal schema (celery tasks). 

73 

74 Args: 

75 data: Data in canonical schema format 

76 

77 Returns: 

78 Data in internal schema format 

79 """ 

80 result = data.copy() 

81 

82 for canonical_field, internal_field in self.FIELD_MAPPINGS.items(): 

83 if canonical_field in result and internal_field not in result: 

84 result[internal_field] = result[canonical_field] 

85 

86 return result 

87 

88 def translate_to_canonical(self, data: dict[str, Any]) -> dict[str, Any]: 

89 """ 

90 Translate from internal schema (celery tasks) to canonical schema (match-scraper). 

91 

92 Args: 

93 data: Data in internal schema format 

94 

95 Returns: 

96 Data in canonical schema format 

97 """ 

98 result = data.copy() 

99 

100 # Reverse mapping 

101 reverse_mappings = {v: k for k, v in self.FIELD_MAPPINGS.items()} 

102 

103 for internal_field, canonical_field in reverse_mappings.items(): 

104 if internal_field in result and canonical_field not in result: 

105 result[canonical_field] = result[internal_field] 

106 

107 return result