11import json
22from pathlib import Path
3- from typing import Any , List , Tuple
43
5- from dataclasses_serialization . json import JSONSerializer
4+ from pydantic import BaseModel , TypeAdapter
65
7- from dmarc_metrics_exporter .dmarc_event import Disposition , Meta
6+ from dmarc_metrics_exporter .dmarc_event import Meta
87
98from .dmarc_metrics import DmarcMetrics , DmarcMetricsCollection , InvalidMeta
109
10+ _Meta = TypeAdapter (Meta )
11+ _DmarcMetrics = TypeAdapter (DmarcMetrics )
12+ _InvalidMeta = TypeAdapter (InvalidMeta )
1113
12- # false positive, pylint: disable=no-value-for-parameter
13- @JSONSerializer .register_serializer (Disposition )
14- def disposition_serializer (disposition : Disposition ) -> str :
15- return disposition .value
1614
17-
18- @JSONSerializer .register_serializer (DmarcMetricsCollection )
19- def dmarc_metrics_collection_serializer (
20- metrics : DmarcMetricsCollection ,
21- ) -> List [Tuple [Any , Any ]]:
22- return JSONSerializer .serialize (
23- {
24- "metrics" : [list (item ) for item in metrics .items ()],
25- "invalid_reports" : [list (item ) for item in metrics .invalid_reports .items ()],
26- }
27- )
28-
29-
30- @JSONSerializer .register_deserializer (Disposition )
31- def disposition_deserializer (_cls , obj : str ) -> Disposition :
32- return Disposition (obj )
33-
34-
35- @JSONSerializer .register_deserializer (DmarcMetricsCollection )
36- def dmarc_metrics_collection_deserializer (_cls , obj ) -> DmarcMetricsCollection :
37- is_old_format = isinstance (obj , list )
38- if is_old_format :
39- obj = {"metrics" : obj }
40- return DmarcMetricsCollection (
41- dict (
42- (
43- JSONSerializer .deserialize (Meta , meta ),
44- JSONSerializer .deserialize (DmarcMetrics , metrics ),
45- )
46- for meta , metrics in obj .get ("metrics" , tuple ())
47- ),
48- dict (
49- (JSONSerializer .deserialize (InvalidMeta , meta ), count )
50- for meta , count in obj .get ("invalid_reports" , tuple ())
51- ),
52- )
15+ class _SerializationModel (BaseModel ):
16+ metrics : list [tuple [Meta , DmarcMetrics ]]
17+ invalid_reports : list [tuple [InvalidMeta , int ]] = []
5318
5419
5520class MetricsPersister :
@@ -59,10 +24,30 @@ def __init__(self, path: Path):
5924 def load (self ) -> DmarcMetricsCollection :
6025 try :
6126 with open (self .path , "r" , encoding = "utf-8" ) as f :
62- return JSONSerializer .deserialize (DmarcMetricsCollection , json .load (f ))
27+ obj = json .load (f )
28+ is_old_format = isinstance (obj , list )
29+ if is_old_format :
30+ obj = {"metrics" : obj }
31+ model = _SerializationModel (** obj )
32+ return DmarcMetricsCollection (
33+ metrics = {
34+ _Meta .validate_python (meta ): _DmarcMetrics .validate_python (
35+ metrics
36+ )
37+ for meta , metrics in model .metrics
38+ },
39+ invalid_reports = {
40+ _InvalidMeta .validate_python (meta ): count
41+ for meta , count in model .invalid_reports
42+ },
43+ )
6344 except FileNotFoundError :
6445 return DmarcMetricsCollection ()
6546
6647 def save (self , metrics : DmarcMetricsCollection ):
48+ model = _SerializationModel (
49+ metrics = [tuple (item ) for item in metrics .items ()],
50+ invalid_reports = [(k , v ) for k , v in metrics .invalid_reports .items ()],
51+ )
6752 with open (self .path , "w" , encoding = "utf-8" ) as f :
68- json . dump ( JSONSerializer . serialize ( metrics ), f )
53+ f . write ( model . model_dump_json () )
0 commit comments