Coverage for functions \ flipdare \ generated \ model \ backend \ app_job_model.py: 89%

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# 

20from __future__ import annotations 

21from datetime import datetime 

22from google.cloud.firestore_v1.transforms import Sentinel 

23from flipdare.core.firestore_field import FirestoreField 

24from flipdare.util.time_util import FirestoreTime 

25from typing import Any, TypedDict, cast, Unpack 

26from enum import StrEnum 

27from pydantic import Field, ConfigDict, TypeAdapter 

28from flipdare.firestore.core.app_base_model import AppBaseModel 

29from flipdare.generated.shared.backend.app_job_type import AppJobType 

30from flipdare.app_types import JsonDict 

31 

32 

33class AppJobKeys(StrEnum): 

34 ID = "id" 

35 CREATED_AT = "created_at" 

36 UPDATED_AT = "updated_at" 

37 JOB_TYPE = "job_type" 

38 OBJ_ID = "obj_id" 

39 PARENT_OBJ_ID = "parent_obj_id" 

40 MODEL_DATA = "model_data" 

41 BEFORE_DATA = "before_data" 

42 MODEL_UPDATES = "model_updates" 

43 VERSION = "version" 

44 PROCESSED = "processed" 

45 ERROR_COUNT = "error_count" 

46 

47 

48# !! IMPORTANT !! 

49# !! 

50# !! this should only be used in the database to query. 

51# !! 

52class AppJobInternalKeys(StrEnum): 

53 CREATED_AT = "created_at" 

54 UPDATED_AT = "updated_at" 

55 VERSION = "VERSION" 

56 PROCESSED = "INT_P" 

57 ERROR_COUNT = "INT_E" 

58 

59 

60class AppJobModel(AppBaseModel): 

61 """Represents a backend scheduled job.""" 

62 

63 model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) 

64 

65 id: str | None = Field(None, alias="id") 

66 created_at: FirestoreField = Field( 

67 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp()) 

68 ) 

69 updated_at: FirestoreField = Field( 

70 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp()) 

71 ) 

72 job_type: AppJobType 

73 obj_id: str 

74 parent_obj_id: str | None = None 

75 model_data: dict[str, Any] | None = None 

76 before_data: dict[str, Any] | None = None 

77 model_updates: dict[str, Any] | None = None 

78 # Version (base internal field) 

79 version: int = Field(default=1, alias="VERSION") 

80 # Processed (base internal field) 

81 processed: bool = Field(default=False, alias="INT_P") 

82 # Error Count (base internal field) 

83 error_count: int = Field(default=0, alias="INT_E") 

84 

85 @classmethod 

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

87 """ 

88 Uses Unpack to give you autocomplete and static warnings 

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

90 

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

92 """ 

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

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

95 if k in cls.__pydantic_fields__: 

96 field_info = cls.__pydantic_fields__[k] 

97 validated_value = cast( 

98 "Any", 

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

100 ) 

101 # Use alias if defined, otherwise use field name 

102 output_key = field_info.alias or k 

103 result[output_key] = validated_value 

104 return result 

105 

106 # ---- Convenience factories ----------------------------------------- 

107 

108 @classmethod 

109 def create_basic( 

110 cls, 

111 job_type: AppJobType, 

112 obj_id: str, 

113 parent_obj_id: str | None = None, 

114 processed: bool = False, 

115 ) -> AppJobModel: 

116 return cls( 

117 id=None, 

118 job_type=job_type, 

119 obj_id=obj_id, 

120 parent_obj_id=parent_obj_id, 

121 INT_P=processed, 

122 ) 

123 

124 @classmethod 

125 def create_new( 

126 cls, 

127 job_type: AppJobType, 

128 obj_id: str, 

129 model: JsonDict, 

130 parent_obj_id: str | None = None, 

131 processed: bool = False, 

132 ) -> AppJobModel: 

133 return cls( 

134 id=None, 

135 job_type=job_type, 

136 obj_id=obj_id, 

137 parent_obj_id=parent_obj_id, 

138 INT_P=processed, 

139 model_data=model, 

140 ) 

141 

142 @classmethod 

143 def create_update( 

144 cls, 

145 job_type: AppJobType, 

146 obj_id: str, 

147 after_model: JsonDict, 

148 before_model: JsonDict, 

149 parent_obj_id: str | None = None, 

150 model_updates: JsonDict | None = None, 

151 processed: bool = False, 

152 ) -> AppJobModel: 

153 before_payload = before_model 

154 after_payload = after_model 

155 

156 return cls( 

157 id=None, 

158 job_type=job_type, 

159 obj_id=obj_id, 

160 parent_obj_id=parent_obj_id, 

161 INT_P=processed, 

162 model_data=after_payload, 

163 before_data=before_payload, 

164 model_updates=model_updates or None, 

165 ) 

166 

167 # ---- Convenience predicates ----------------------------------------- 

168 

169 @property 

170 def is_update(self) -> bool: 

171 return self.model_updates is None 

172 

173 

174APPJOB_FIELD_NAMES: list[str] = list(AppJobModel.model_fields.keys()) 

175 

176 

177class AppJobDict(TypedDict, total=False): 

178 id: str | None 

179 created_at: Sentinel | datetime | str 

180 updated_at: Sentinel | datetime | str 

181 job_type: AppJobType 

182 obj_id: str 

183 parent_obj_id: str | None 

184 model_data: dict[str, Any] | None 

185 before_data: dict[str, Any] | None 

186 model_updates: dict[str, Any] | None 

187 VERSION: int | None 

188 INT_P: bool | None 

189 INT_E: int | None