Coverage for functions \ flipdare \ generated \ model \ backend \ app_log_model.py: 86%
136 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#
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.system_log_type import SystemLogType
30from flipdare.generated.shared.app_log_category import AppLogCategory
31from flipdare.generated.shared.firestore_collections import FirestoreCollections
32from flipdare.generated.shared.search.search_collections import SearchCollections
33from flipdare.generated.shared.backend.app_job_type import AppJobType
34from flipdare.constants import NO_DOC_ID
35from pydantic import field_validator
36from flipdare.error.app_error_protocol import AppErrorProtocol
37from flipdare.generated.shared.app_payment_error_code import AppPaymentErrorCode
38from flipdare.generated.shared.app_error_code import AppErrorCode
39from flipdare.util.time_util import TimeUtil
42class AppLogKeys(StrEnum):
43 ID = "id"
44 CREATED_AT = "created_at"
45 UPDATED_AT = "updated_at"
46 LOG_TYPE = "log_type"
47 CATEGORY = "category"
48 ERROR_CODE = "error_code"
49 SOURCE = "source"
50 CALLED_BY = "called_by"
51 FIRESTORE_COLLECTION = "firestore_collection"
52 SEARCH_COLLECTION = "search_collection"
53 JOB_TYPE = "job_type"
54 ADMIN_NOTIFIED = "admin_notified"
55 ACKNOWLEDGED = "acknowledged"
56 MESSAGE = "message"
57 OBJ_ID = "obj_id"
58 EXTRA = "extra"
59 STACK_TRACE = "stack_trace"
60 VERSION = "version"
61 PROCESSED = "processed"
64# !! IMPORTANT !!
65# !!
66# !! this should only be used in the database to query.
67# !!
68class AppLogInternalKeys(StrEnum):
69 CREATED_AT = "created_at"
70 UPDATED_AT = "updated_at"
71 VERSION = "VERSION"
72 PROCESSED = "INT_P"
75class AppLogModel(AppBaseModel):
76 """Represents a backend log entry."""
78 model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)
80 id: str | None = Field(None, alias="id")
81 created_at: FirestoreField = Field(
82 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp())
83 )
84 updated_at: FirestoreField = Field(
85 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp())
86 )
87 log_type: SystemLogType
88 category: AppLogCategory
89 error_code: AppErrorProtocol | None = None
90 source: str
91 called_by: str | None = None
92 firestore_collection: FirestoreCollections | None = None
93 search_collection: SearchCollections | None = None
94 job_type: AppJobType | None = None
95 admin_notified: bool
96 acknowledged: bool
97 message: str
98 obj_id: str | None = None
99 extra: dict[str, Any] | None = None
100 stack_trace: str | None = None
101 # Version (base internal field)
102 version: int = Field(default=1, alias="VERSION")
103 # Processed (base internal field)
104 processed: bool = Field(default=False, alias="INT_P")
106 @classmethod
107 def validate_partial(cls, **data: Unpack[AppLogDict]) -> dict[str, Any]:
108 """
109 Uses Unpack to give you autocomplete and static warnings
110 if you pass an invalid key or type in your code.
112 Returns a dict with Firestore field names (aliases) for use with batch.update().
113 """
114 result: dict[str, Any] = {}
115 for k, v in data.items():
116 if k in cls.__pydantic_fields__:
117 field_info = cls.__pydantic_fields__[k]
118 validated_value = cast(
119 "Any",
120 TypeAdapter(field_info.annotation).validate_python(v),
121 )
122 # Use alias if defined, otherwise use field name
123 output_key = field_info.alias or k
124 result[output_key] = validated_value
125 return result
127 # ---- BaseModel validators ------------------------------------------
128 @field_validator("error_code", mode="before")
129 @classmethod
130 def transform_to_enum(cls, v: Any) -> Any:
131 if isinstance(v, str):
132 # Try to find which Enum this string belongs to
133 for enum_cls in [AppErrorCode, AppPaymentErrorCode]:
134 try:
135 return enum_cls(v)
136 except ValueError:
137 continue
138 return v
140 # ---- Convenience factories -----------------------------------------
142 @classmethod
143 def system(
144 cls,
145 error_code: AppErrorProtocol,
146 obj_id: str,
147 message: str,
148 called_by: str,
149 admin_notified: bool,
150 job_type: AppJobType | None = None,
151 extra: dict[str, Any] | None = None,
152 source: str | None = None,
153 stack_trace: str | None = None,
154 log_type: SystemLogType = SystemLogType.ERROR,
155 ) -> AppLogModel:
157 return cls(
158 id=None,
159 log_type=log_type,
160 category=error_code.category,
161 message=message,
162 called_by=called_by,
163 acknowledged=False,
164 admin_notified=admin_notified,
165 job_type=job_type,
166 error_code=error_code,
167 obj_id=obj_id,
168 source=cls.get_actual_source(source, job_type, called_by),
169 extra=cls._validate_extra(extra),
170 stack_trace=stack_trace,
171 )
173 @classmethod
174 def job(
175 cls,
176 called_by: str,
177 job_type: AppJobType,
178 obj_id: str,
179 message: str,
180 error_code: AppErrorProtocol,
181 source: str,
182 extra: dict[str, Any],
183 admin_notified: bool = False,
184 log_type: SystemLogType = SystemLogType.ERROR,
185 stack_trace: str | None = None,
186 ) -> AppLogModel:
187 return cls(
188 id=None,
189 log_type=log_type,
190 category=job_type.category,
191 message=message,
192 called_by=called_by,
193 admin_notified=admin_notified,
194 acknowledged=False,
195 job_type=job_type,
196 error_code=error_code,
197 obj_id=obj_id,
198 source=cls.get_actual_source(source, job_type, called_by),
199 extra=cls._validate_extra(extra),
200 stack_trace=stack_trace,
201 )
203 @classmethod
204 def search(
205 cls,
206 called_by: str,
207 job_type: AppJobType,
208 error_code: AppErrorProtocol,
209 obj_id: str,
210 message: str,
211 source: str,
212 extra: dict[str, Any],
213 admin_notified: bool = False,
214 log_type: SystemLogType = SystemLogType.ERROR,
215 stack_trace: str | None = None,
216 ) -> AppLogModel:
217 return cls(
218 id=None,
219 log_type=log_type,
220 category=error_code.category,
221 message=message,
222 called_by=called_by,
223 admin_notified=admin_notified,
224 acknowledged=False,
225 job_type=job_type,
226 error_code=error_code,
227 obj_id=obj_id,
228 source=source,
229 extra=cls._validate_extra(extra),
230 stack_trace=stack_trace,
231 )
233 @classmethod
234 def info(
235 cls,
236 message: str,
237 called_by: str,
238 job_type: AppJobType | None = None,
239 obj_id: str | None = None,
240 source: str | None = None,
241 extra: dict[str, Any] | None = None,
242 ) -> AppLogModel:
243 if obj_id is None:
244 obj_id = NO_DOC_ID
246 return cls(
247 id=None,
248 log_type=SystemLogType.INFO,
249 category=job_type.category if job_type else AppLogCategory.OTHER,
250 message=message,
251 called_by=called_by,
252 admin_notified=False,
253 acknowledged=True,
254 job_type=job_type,
255 obj_id=obj_id,
256 source=cls.get_actual_source(source, job_type, called_by),
257 extra=cls._validate_extra(extra),
258 )
260 # ---- Convenience predicates -----------------------------------------
262 @staticmethod
263 def get_actual_source(
264 source: str | None, job_type: AppJobType | None, called_by: str | None = None
265 ) -> str:
266 if source is not None:
267 return source
268 if job_type is not None:
269 return job_type.value
270 if called_by is not None:
271 return called_by
273 return "NoSource"
275 @staticmethod
276 def _validate_extra(extra: dict[str, Any] | None) -> dict[str, Any] | None:
277 if extra is None:
278 return None
280 # additional check, if any value is a Sentinel, we should replace it with
281 # the current timestamp in ISO format
282 now_iso = FirestoreTime.formatted(TimeUtil.get_current_utc_dt())
283 return {
284 key: (now_iso if (key.endswith("_at") and not isinstance(value, Sentinel)) else value)
285 for key, value in extra.items()
286 }
289APPLOG_FIELD_NAMES: list[str] = list(AppLogModel.model_fields.keys())
292class AppLogDict(TypedDict, total=False):
293 id: str | None
294 created_at: Sentinel | datetime | str
295 updated_at: Sentinel | datetime | str
296 log_type: SystemLogType
297 category: AppLogCategory
298 error_code: AppErrorProtocol | None
299 source: str
300 called_by: str | None
301 firestore_collection: FirestoreCollections | None
302 search_collection: SearchCollections | None
303 job_type: AppJobType | None
304 admin_notified: bool
305 acknowledged: bool
306 message: str
307 obj_id: str | None
308 extra: dict[str, Any] | None
309 stack_trace: str | None
310 VERSION: int | None
311 INT_P: bool | None