Coverage for functions \ flipdare \ task \ report \ core \ query_report.py: 81%

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 collections.abc import Callable, Mapping, Sequence 

15from typing import Any, override 

16from flipdare.backend.app_logger import AppLogger 

17from flipdare.app_log import LOG 

18from flipdare.constants import IS_DEBUG 

19from flipdare.mailer.admin_mailer import AdminMailer 

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

21from flipdare.task.report.core.table_report import TableReport 

22from flipdare.wrapper import PersistedWrapper 

23 

24 

25class QueryReport[TSchema: Mapping[str, Any]](TableReport[TSchema]): 

26 __slots__ = ("_query_fn",) 

27 

28 def __init__( 

29 self, 

30 job_type: AppJobType, 

31 schema_cls: type[TSchema], 

32 query_fn: Callable[..., Sequence[PersistedWrapper[Any]]], 

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

34 app_logger: AppLogger, 

35 mailer: AdminMailer, 

36 ) -> None: 

37 super().__init__( 

38 job_type=job_type, 

39 schema_cls=schema_cls, 

40 process_fn=process_fn, 

41 app_logger=app_logger, 

42 mailer=mailer, 

43 ) 

44 self._query_fn = query_fn 

45 

46 @property 

47 def query_fn(self) -> Callable[..., Sequence[PersistedWrapper[Any]]]: 

48 return self._query_fn 

49 

50 @property 

51 def process_fn(self) -> Callable[..., TSchema]: 

52 return self._process_fn 

53 

54 @override 

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

56 try: 

57 results = self._query_fn() 

58 total_ct = len(results) 

59 if IS_DEBUG: 

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

61 return results 

62 

63 except Exception as ex: 

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

65 self.add_error(msg) 

66 return None