Coverage for polar/metrics/schemas.py: 71%

40 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 15:52 +0000

1from datetime import date 1a

2from typing import TYPE_CHECKING 1a

3 

4from pydantic import AwareDatetime, Field, create_model 1a

5 

6from polar.kit.schemas import Schema 1a

7 

8from .metrics import METRICS, MetricType 1a

9 

10 

11class Metric(Schema): 1a

12 """Information about a metric.""" 

13 

14 slug: str = Field(description="Unique identifier for the metric.") 1a

15 display_name: str = Field(description="Human-readable name for the metric.") 1a

16 type: MetricType = Field( 1a

17 description=( 

18 "Type of the metric, useful to know the unit or format of the value." 

19 ) 

20 ) 

21 

22 

23if TYPE_CHECKING: 23 ↛ 25line 23 didn't jump to line 25 because the condition on line 23 was never true1a

24 

25 class Metrics(Schema): 

26 def __getattr__(self, name: str) -> Metric: ... 

27 

28else: 

29 Metrics = create_model( 1a

30 "Metrics", **{m.slug: (Metric, ...) for m in METRICS}, __base__=Schema 

31 ) 

32 

33 

34class MetricsPeriodBase(Schema): 1a

35 """ 

36 A period of time with metrics data. 

37 

38 It maps each metric slug to its value for this timestamp. 

39 """ 

40 

41 timestamp: AwareDatetime = Field(description="Timestamp of this period data.") 1a

42 

43 

44if TYPE_CHECKING: 44 ↛ 46line 44 didn't jump to line 46 because the condition on line 44 was never true1a

45 

46 class MetricsPeriod(MetricsPeriodBase): 

47 def __getattr__(self, name: str) -> int | float: ... 

48 

49else: 

50 MetricsPeriod = create_model( 1a

51 "MetricPeriod", 

52 **{m.slug: (int | float, ...) for m in METRICS}, 

53 __base__=MetricsPeriodBase, 

54 ) 

55 

56 

57class MetricsTotalsBase(Schema): 1a

58 """ 

59 Metrics totals over the whole selected period. 

60 

61 It maps each metric slug to its value for this period. The aggregation is done 

62 differently depending on the metric type. 

63 """ 

64 

65 

66if TYPE_CHECKING: 66 ↛ 68line 66 didn't jump to line 68 because the condition on line 66 was never true1a

67 

68 class MetricsTotals(MetricsTotalsBase): 

69 def __getattr__(self, name: str) -> int | float: ... 

70 

71 

72else: 

73 MetricsTotals = create_model( 1a

74 "MetricsTotals", 

75 **{m.slug: (int | float, ...) for m in METRICS}, 

76 __base__=MetricsTotalsBase, 

77 ) 

78 

79 

80class MetricsResponse(Schema): 1a

81 """Metrics response schema.""" 

82 

83 periods: list[MetricsPeriod] = Field(description="List of data for each timestamp.") 1a

84 totals: MetricsTotals = Field(description="Totals for the whole selected period.") 1a

85 metrics: Metrics = Field(description="Information about the returned metrics.") 1a

86 

87 

88class MetricsIntervalLimit(Schema): 1a

89 """Date interval limit to get metrics for a given interval.""" 

90 

91 min_days: int = Field(description="Minimum number of days for this interval.") 1a

92 max_days: int = Field(description="Maximum number of days for this interval.") 1a

93 

94 

95class MetricsIntervalsLimits(Schema): 1a

96 """Date interval limits to get metrics for each interval.""" 

97 

98 hour: MetricsIntervalLimit = Field(description="Limits for the hour interval.") 1a

99 day: MetricsIntervalLimit = Field(description="Limits for the day interval.") 1a

100 week: MetricsIntervalLimit = Field(description="Limits for the week interval.") 1a

101 month: MetricsIntervalLimit = Field(description="Limits for the month interval.") 1a

102 year: MetricsIntervalLimit = Field(description="Limits for the year interval.") 1a

103 

104 

105class MetricsLimits(Schema): 1a

106 """Date limits to get metrics.""" 

107 

108 min_date: date = Field(description="Minimum date to get metrics.") 1a

109 intervals: MetricsIntervalsLimits = Field(description="Limits for each interval.") 1a