Coverage for functions \ flipdare \ firestore \ core \ collection_stat_query.py: 100%

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 

14from google.cloud.firestore import And 

15from datetime import datetime 

16from typing import Any 

17from flipdare.firestore.core.db_query import DateFilter, DbQuery, FieldOp, WhereField 

18 

19_processed_key: str = "INT_P" # processed: bool 

20_error_key: str = "INT_E" # error_count: int 

21 

22_OP = FieldOp 

23 

24 

25class CollectionStatQuery(DbQuery): 

26 

27 def __init__( 

28 self, 

29 where_fields: list[WhereField[Any]], 

30 from_date: datetime, 

31 to_date: datetime, 

32 sum_field: str | None = None, 

33 ) -> None: 

34 filter_values = [field.filter for field in where_fields] 

35 filter_values.extend(DateFilter.created_at(from_date=from_date, to_date=to_date).filters) 

36 

37 filter_value = And(filter_values) # type: ignore[arg-type] 

38 super().__init__(filter_value=filter_value) 

39 self.sum_field = sum_field 

40 

41 @classmethod 

42 def processed(cls, from_date: datetime, to_date: datetime) -> CollectionStatQuery: 

43 return cls( 

44 where_fields=[WhereField[Any](_processed_key, _OP.EQUAL, True)], 

45 from_date=from_date, 

46 to_date=to_date, 

47 ) 

48 

49 @classmethod 

50 def unprocessed(cls, from_date: datetime, to_date: datetime) -> CollectionStatQuery: 

51 return cls( 

52 where_fields=[WhereField[Any](_processed_key, _OP.EQUAL, False)], 

53 from_date=from_date, 

54 to_date=to_date, 

55 ) 

56 

57 @classmethod 

58 def error(cls, from_date: datetime, to_date: datetime) -> CollectionStatQuery: 

59 return cls( 

60 where_fields=[WhereField[Any](_error_key, _OP.GREATER_THAN, 0)], 

61 from_date=from_date, 

62 to_date=to_date, 

63 sum_field=_error_key, 

64 ) 

65 

66 @classmethod 

67 def custom( 

68 cls, 

69 from_date: datetime, 

70 to_date: datetime, 

71 where_fields: list[WhereField[Any]], 

72 sum_field: str | None = None, 

73 ) -> CollectionStatQuery: 

74 return cls( 

75 where_fields=where_fields, 

76 from_date=from_date, 

77 to_date=to_date, 

78 sum_field=sum_field, 

79 )