Coverage for functions \ flipdare \ backend \ job_logger.py: 64%

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# 

12 

13from __future__ import annotations 

14 

15from flipdare.app_types import DatabaseDict 

16from flipdare.core.singleton import Singleton 

17from flipdare.firestore.backend.app_job_db import AppJobDb 

18from flipdare.generated import AppJobModel, AppJobType 

19from flipdare.generated.shared.app_error_code import AppErrorCode 

20from flipdare.wrapper import AppJobWrapper 

21 

22__all__ = ["JobLogger"] 

23 

24 

25class JobLogger(Singleton): 

26 

27 def __init__(self, job_db: AppJobDb | None = None) -> None: 

28 self._job_db = job_db 

29 

30 @property 

31 def job_db(self) -> AppJobDb: 

32 from flipdare.services import get_job_db 

33 

34 if self._job_db is None: 

35 self._job_db = get_job_db() 

36 return self._job_db 

37 

38 def get_jobs(self, task_types: list[AppJobType]) -> list[AppJobWrapper] | None: 

39 job_db = self.job_db 

40 try: 

41 return job_db.get_jobs_to_process(job_types=task_types) 

42 except Exception as error: 

43 msg = f"Error getting dare tasks: {error}" 

44 self._send_error(msg) 

45 return None 

46 

47 def resolve_jobs(self, jobs: list[AppJobWrapper]) -> None: 

48 job_ids: list[str] = [] 

49 for job in jobs: 

50 doc_id = job.doc_id 

51 job_ids.append(doc_id) 

52 

53 self.job_db.resolve_jobs(job_ids=job_ids) 

54 

55 def create_basic(self, doc_id: str, job_type: AppJobType) -> None: 

56 job = AppJobModel.create_basic(job_type=job_type, obj_id=doc_id) 

57 self._create(job) 

58 

59 def create_new( 

60 self, 

61 doc_id: str, 

62 job_type: AppJobType, 

63 model: DatabaseDict, 

64 ) -> AppJobWrapper | None: 

65 job = AppJobModel.create_new( 

66 job_type=job_type, 

67 obj_id=doc_id, 

68 model=model, 

69 ) 

70 return self._create(job) 

71 

72 def create_update( 

73 self, 

74 doc_id: str, 

75 job_type: AppJobType, 

76 before_model: DatabaseDict, 

77 after_model: DatabaseDict, 

78 updated_data: DatabaseDict | None = None, 

79 ) -> AppJobWrapper | None: 

80 job = AppJobModel.create_update( 

81 job_type=job_type, 

82 obj_id=doc_id, 

83 after_model=after_model, 

84 before_model=before_model, 

85 model_updates=updated_data, 

86 ) 

87 return self._create(job) 

88 

89 def _create(self, job: AppJobModel) -> AppJobWrapper | None: 

90 try: 

91 return self.job_db.create(job) 

92 except Exception as e: 

93 msg = f"Failed to create job for {job.job_type.value}: {job.obj_id}: {e!s}" 

94 self._send_error(msg, job_type=job.job_type) 

95 return None 

96 

97 def _send_error(self, msg: str, job_type: AppJobType | None = None) -> None: 

98 from flipdare.mailer.admin_mailer import AdminMailer 

99 

100 AdminMailer.instance().send_error( 

101 error_code=AppErrorCode.TRIGGER, 

102 job_type=job_type, 

103 message=msg, 

104 )