(
self,
*,
stage: str,
status: str,
message: str,
account_id: str,
run_id: str,
action: str,
source: str,
product_id: str,
pay_type: str,
details: dict[str, Any] | None,
)
| 228 | self.logger.log(level, "[system][%s] %s", stage, message) |
| 229 | |
| 230 | def _base_entry( |
| 231 | self, |
| 232 | *, |
| 233 | stage: str, |
| 234 | status: str, |
| 235 | message: str, |
| 236 | account_id: str, |
| 237 | run_id: str, |
| 238 | action: str, |
| 239 | source: str, |
| 240 | product_id: str, |
| 241 | pay_type: str, |
| 242 | details: dict[str, Any] | None, |
| 243 | ) -> dict[str, Any]: |
| 244 | now = datetime.now().astimezone() |
| 245 | return { |
| 246 | "timestamp": now.isoformat(timespec="seconds"), |
| 247 | "date": now.strftime("%Y-%m-%d"), |
| 248 | "pid": os.getpid(), |
| 249 | "thread": threading.current_thread().name, |
| 250 | "account_id": account_id, |
| 251 | "run_id": run_id, |
| 252 | "action": action, |
| 253 | "source": source, |
| 254 | "stage": stage, |
| 255 | "status": status, |
| 256 | "product_id": product_id, |
| 257 | "pay_type": pay_type, |
| 258 | "message": message, |
| 259 | "details": self._sanitize_value(details or {}), |
| 260 | } |
| 261 | |
| 262 | def _write_entry(self, entry: dict[str, Any]) -> None: |
| 263 | date_part = str(entry.get("date") or datetime.now().strftime("%Y-%m-%d")) |
no test coverage detected