Coverage for functions \ flipdare \ task \ report \ core \ table_report.py: 84%

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 

13 

14from abc import ABC, abstractmethod 

15from collections.abc import Callable, Mapping, Sequence 

16from typing import Any, override 

17from flipdare.task.report.core._admin_report import AdminReport 

18from flipdare.backend.app_logger import AppLogger 

19from flipdare.app_log import LOG 

20from flipdare.app_types import ReportListType 

21from flipdare.constants import EMAIL_MAX_LINE_LENGTH, IS_DEBUG 

22from flipdare.mailer.core.email_composer import EmailComposer 

23from flipdare.mailer.admin_mailer import AdminMailer 

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

25 

26 

27class TableReport[TSchema: Mapping[str, Any]](AdminReport, ABC): 

28 __slots__ = ( 

29 "_process_fn", 

30 "_schema_cls", 

31 ) 

32 

33 def __init__( 

34 self, 

35 job_type: AppJobType, 

36 schema_cls: type[TSchema], 

37 process_fn: Callable[..., TSchema], 

38 app_logger: AppLogger, 

39 mailer: AdminMailer, 

40 ) -> None: 

41 super().__init__( 

42 job_type=job_type, 

43 app_logger=app_logger, 

44 mailer=mailer, 

45 ) 

46 self._schema_cls = schema_cls 

47 self._process_fn = process_fn 

48 

49 @abstractmethod 

50 def table_data(self) -> Sequence[Any] | None: ... 

51 

52 @property 

53 def keys(self) -> list[str]: 

54 return list(self._schema_cls.__annotations__.keys()) 

55 

56 @property 

57 def headers(self) -> list[str]: 

58 return [key.replace("_", " ").title() for key in self.keys] 

59 

60 @override 

61 def compose(self) -> EmailComposer | None: 

62 report_data: ReportListType = [] 

63 

64 try: 

65 results = self.table_data() 

66 if results is None: 

67 msg = f"Report {self.job_type.value} returned no data" 

68 self.add_error(msg) 

69 return None 

70 

71 total_ct = len(results) 

72 if IS_DEBUG: 

73 LOG().debug(f"Report {self.job_type.value} generated {total_ct} results") 

74 

75 for result in results: 

76 try: 

77 entry = self._process_fn(result) 

78 raw_keys = self.keys 

79 formatted_values = [str(entry.get(k, "")) for k in raw_keys] 

80 report_data.append(formatted_values) 

81 except Exception as ex: 

82 # since we send errors we need to ensure 

83 # 500, b'Line too long (see RFC5321 4.5.3.1.6)'): 

84 

85 if hasattr(result, "doc_id"): 

86 identifier = getattr(result, "doc_id", "unknown") 

87 else: 

88 identifier = str(result) 

89 

90 error_msg = f"Failed to generate report entry for doc {identifier}: {ex}" 

91 error_msg = ( 

92 error_msg[:EMAIL_MAX_LINE_LENGTH] + "..." 

93 if len(error_msg) > EMAIL_MAX_LINE_LENGTH 

94 else error_msg 

95 ) 

96 self.add_error(error_msg) 

97 

98 except Exception as ex: 

99 msg = f"Failed to generate report data: {ex}" 

100 self.add_error(msg) 

101 return None 

102 

103 try: 

104 return EmailComposer.table( 

105 data=report_data, 

106 headers=self.headers, 

107 priority=self.priority, 

108 ) 

109 except Exception as ex: 

110 msg = f"Failed to generate report for {self._debug_label}\tError: {ex}" 

111 self.add_error(msg) 

112 return None