Coverage for functions \ flipdare \ backend \ app_stats.py: 70%
89 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
13from flipdare.backend.app_logger import AppLogger
14from flipdare.app_log import LOG
15from flipdare.app_types import CronResult
16from flipdare.constants import IS_TRACE
17from flipdare.result.output_result import OutputResult
18from flipdare.core.singleton import Singleton
19from flipdare.firestore.backend.app_stat_db import AppStatDb
20from flipdare.generated.model.backend.app_stat_metric_model import AppStatMetricModel
21from flipdare.generated.model.backend.metric.count_metric import CountMetric
22from flipdare.generated.model.backend.metric.outcome_metric import OutcomeMetric
23from flipdare.generated.shared.app_error_code import AppErrorCode
24from flipdare.generated.shared.backend.app_job_type import AppJobType
25from flipdare.generated.shared.firestore_collections import FirestoreCollections
27__all__ = ["AppStats"]
30class AppStats(Singleton):
31 """
32 Performance statistics admin.
33 """
35 def __init__(
36 self,
37 stat_db: AppStatDb | None = None,
38 app_logger: AppLogger | None = None,
39 ) -> None:
40 super().__init__()
41 self._stat_db = stat_db
42 self._app_logger = app_logger
44 @property
45 def stat_db(self) -> AppStatDb:
46 from flipdare.services import get_db_manager
48 if self._stat_db is None:
49 self._stat_db = get_db_manager().stat_db
50 return self._stat_db
52 @property
53 def app_logger(self) -> AppLogger:
54 from flipdare.services import get_app_logger
56 if self._app_logger is None:
57 self._app_logger = get_app_logger()
58 return self._app_logger
60 def add(
61 self,
62 job_type: AppJobType,
63 result: CronResult,
64 duration: int | None = None,
65 ) -> None:
66 match result:
67 case CountMetric():
68 self.add_count_metric(job_type=job_type, metric=result)
69 case OutcomeMetric():
70 self.add_outcome_metric(job_type=job_type, metric=result)
71 case OutputResult():
72 self.add_result(job_type=job_type, output_result=result, duration=duration)
74 def add_result(
75 self,
76 job_type: AppJobType,
77 output_result: OutputResult,
78 duration: int | None = None,
79 ) -> None:
80 succeeded = output_result.is_ok
81 duration = duration or output_result.duration
83 if IS_TRACE:
84 msg = (
85 f"Logging output result for {job_type.label}: "
86 f"success={succeeded}, "
87 f"message={output_result.message}, "
88 f"duration={duration}s"
89 )
90 LOG().trace(msg)
92 self.add_outcome_metric(
93 job_type,
94 OutcomeMetric(
95 succeeded=succeeded,
96 duration=duration,
97 ),
98 )
100 def add_outcome(
101 self,
102 job_type: AppJobType,
103 succeeded: bool,
104 duration: int,
105 ) -> None:
106 if IS_TRACE:
107 msg = (
108 f"Logging outcome metric for {job_type.label}: "
109 f"succeeded={succeeded}, duration={duration}s"
110 )
111 LOG().trace(msg)
113 self.add_outcome_metric(
114 job_type,
115 OutcomeMetric(
116 succeeded=succeeded,
117 duration=duration,
118 ),
119 )
121 def add_outcome_metric(
122 self,
123 job_type: AppJobType,
124 metric: OutcomeMetric,
125 ) -> None:
126 if IS_TRACE:
127 msg = f"Logging outcome metric for {job_type.label}: {metric}"
128 LOG().trace(msg)
130 model: AppStatMetricModel | None = None
131 try:
132 model = AppStatMetricModel(
133 id=None,
134 job_type=job_type,
135 metric=metric,
136 )
137 self._add_stat(job_type, model)
138 except Exception as ex:
139 # this is most likely a pydantic validation error..
140 cause = f"Failed to create outcome metric for {job_type.label}: {ex}"
141 LOG().error(cause)
142 self.app_logger.unexpected_code_path(
143 job_type=job_type,
144 collection=FirestoreCollections.APP_STAT_METRIC,
145 message=cause,
146 ex_error=ex,
147 )
148 return
150 def add_count(
151 self,
152 job_type: AppJobType,
153 success_ct: int,
154 failed_ct: int,
155 skipped_ct: int,
156 duration: int,
157 ) -> None:
158 if IS_TRACE:
159 msg = (
160 f"Logging count metric for {job_type.label}: "
161 f"success={success_ct}, failed={failed_ct}, skipped={skipped_ct}, duration={duration}s"
162 )
163 LOG().trace(msg)
165 self.add_count_metric(
166 job_type,
167 CountMetric(
168 success_ct=success_ct,
169 failed_ct=failed_ct,
170 skipped_ct=skipped_ct,
171 duration=duration,
172 ),
173 )
175 def add_count_metric(
176 self,
177 job_type: AppJobType,
178 metric: CountMetric,
179 ) -> None:
180 if IS_TRACE:
181 msg = f"Logging count metric for {job_type.label}: {metric}"
182 LOG().trace(msg)
184 model: AppStatMetricModel | None = None
185 try:
186 model = AppStatMetricModel(
187 id=None,
188 job_type=job_type,
189 metric=metric,
190 )
191 self._add_stat(job_type, model)
192 except Exception as ex:
193 # this is most likely a pydantic validation error..
194 cause = f"Failed to create count metric for {job_type.label}: {ex}"
195 LOG().error(cause)
196 self.app_logger.unexpected_code_path(
197 job_type=job_type,
198 collection=FirestoreCollections.APP_STAT_METRIC,
199 message=cause,
200 ex_error=ex,
201 )
202 return
204 def _add_stat(
205 self,
206 job_type: AppJobType,
207 metric: AppStatMetricModel,
208 ) -> None:
209 try:
210 self.stat_db.add(metric=metric)
211 except Exception as ex:
212 cause = f"Failed to log count metric for {job_type.label}: {ex}"
213 LOG().error(cause)
214 self.app_logger.db_error(
215 error_code=AppErrorCode.DATABASE_EX,
216 job_type=job_type,
217 collection=FirestoreCollections.APP_STAT_METRIC,
218 message=cause,
219 ex_error=ex,
220 data=metric.to_dict(),
221 )