Coverage for functions \ flipdare \ generated \ model \ internal \ stopwatch_model.py: 100%

0 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-05-08 12:22 +1000

1#!/usr/bin/env python 

2# 

3# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

4# 

5# This file is part of Flipdare's proprietary software and contains 

6# confidential and copyrighted material. Unauthorised copying, 

7# modification, distribution, or use of this file is strictly 

8# prohibited without prior written permission from Flipdare Pty Ltd. 

9# 

10# This software includes third-party components licensed under MIT, 

11# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

12# 

13# NOTE: THIS FILE IS AUTO GENERATED. DO NOT EDIT. 

14# 

15# Generated by codegen_models.py 

16# 

17# Modify 'codegen_models.py' 

18# and re-run the script above to update. 

19# 

20# pragma: no cover 

21from __future__ import annotations 

22from typing import Any, TypedDict, cast, Unpack 

23from enum import StrEnum 

24from pydantic import Field, ConfigDict, TypeAdapter 

25from flipdare.firestore.core.app_base_model import AppBaseModel 

26from flipdare.generated.shared.model.core.stopwatch_duration import StopwatchDuration 

27from flipdare.util.time_util import TimeUtil 

28import math 

29from datetime import datetime 

30from pydantic import model_validator 

31 

32 

33class StopwatchKeys(StrEnum): 

34 STARTED_AT = "started_at" 

35 EXPIRES_AT = "expires_at" 

36 DURATION = "duration" 

37 

38 

39class StopwatchModel(AppBaseModel): 

40 """Stores a time""" 

41 

42 model_config = ConfigDict(populate_by_name=True) 

43 

44 started_at: float = Field(default=0.0) 

45 expires_at: float = Field(default=0.0) 

46 duration: StopwatchDuration 

47 

48 @classmethod 

49 def validate_partial(cls, **data: Unpack[StopwatchDict]) -> dict[str, Any]: 

50 """ 

51 Uses Unpack to give you autocomplete and static warnings 

52 if you pass an invalid key or type in your code. 

53 

54 Returns a dict with Firestore field names (aliases) for use with batch.update(). 

55 """ 

56 result: dict[str, Any] = {} 

57 for k, v in data.items(): 

58 if k in cls.__pydantic_fields__: 

59 field_info = cls.__pydantic_fields__[k] 

60 validated_value = cast( 

61 "Any", 

62 TypeAdapter(field_info.annotation).validate_python(v), 

63 ) 

64 # Use alias if defined, otherwise use field name 

65 output_key = field_info.alias or k 

66 result[output_key] = validated_value 

67 return result 

68 

69 # ---- Convenience classmethods ----------------------------------------- 

70 

71 @classmethod 

72 def from_start(cls, started_at: datetime, duration: StopwatchDuration) -> StopwatchModel: 

73 expires_at = TimeUtil.get_utc_time_future_days(started_at, duration.days) 

74 return cls( 

75 started_at=started_at.timestamp(), 

76 expires_at=expires_at.timestamp(), 

77 duration=duration, 

78 ) 

79 

80 @classmethod 

81 def from_now(cls, duration: StopwatchDuration) -> StopwatchModel: 

82 now = TimeUtil.get_current_utc_dt() 

83 return cls.from_start(started_at=now, duration=duration) 

84 

85 # ---- Validation ----------------------------------------- 

86 

87 @model_validator(mode="after") 

88 def calculate_expiration(self) -> StopwatchModel: 

89 from flipdare.app_log import LOG 

90 

91 # 1. Determine started_dt 

92 if self.started_at: # Handles both float > 0 and datetime objects 

93 # Ensure we have a datetime object to do math 

94 started_dt = TimeUtil.epoch_to_utc_dt(self.started_at) 

95 else: 

96 started_dt = TimeUtil.get_current_utc_dt() 

97 

98 # 2. Calculate expected expiry 

99 expected_expires_dt = TimeUtil.get_utc_time_future_days(started_dt, self.duration.days) 

100 expected_timestamp = expected_expires_dt.timestamp() 

101 

102 # 3. Check if provided expires_at matches (with a tiny buffer for float math) 

103 is_close = math.isclose(self.expires_at, expected_timestamp, abs_tol=0.001) 

104 if self.expires_at and not is_close: 

105 msg = f"Invalid expires_at: expected={expected_timestamp}, got={self.expires_at}. Resetting." 

106 LOG().warning(msg) 

107 

108 # 4. Update the instance 

109 self.started_at = started_dt.timestamp() 

110 self.expires_at = expected_timestamp 

111 

112 return self 

113 

114 # ---- Convenience predicates ----------------------------------------- 

115 

116 @property 

117 def is_expired(self) -> bool: 

118 now_ts = TimeUtil.get_current_utc_dt().timestamp() 

119 return now_ts >= self.expires_at 

120 

121 @property 

122 def started_at_dt(self) -> datetime: 

123 return TimeUtil.epoch_to_utc_dt(self.started_at) 

124 

125 @property 

126 def expires_at_dt(self) -> datetime: 

127 return TimeUtil.epoch_to_utc_dt(self.expires_at) 

128 

129 def started_at_formatted(self, local_tz: str | None = None) -> str: 

130 return TimeUtil.formatted_user(self.started_at_dt, user_tz_str=local_tz) 

131 

132 def expires_at_formatted(self, local_tz: str | None = None) -> str: 

133 return TimeUtil.formatted_user(self.expires_at_dt, user_tz_str=local_tz) 

134 

135 

136STOPWATCH_FIELD_NAMES: list[str] = list(StopwatchModel.model_fields.keys()) 

137 

138 

139class StopwatchDict(TypedDict, total=False): 

140 started_at: float | None 

141 expires_at: float | None 

142 duration: StopwatchDuration