Coverage for functions \ flipdare \ core \ firestore_field.py: 97%

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 datetime import UTC, datetime 

15from typing import Annotated, Any 

16 

17from google.cloud.firestore_v1.transforms import Sentinel 

18 

19from pydantic import SerializationInfo # Import this to check the mode 

20from pydantic import ( 

21 BeforeValidator, 

22 PlainSerializer, 

23 WithJsonSchema, 

24) 

25 

26from flipdare.util.time_util import FirestoreTime 

27 

28 

29def validate_firestore_obj(v: Any) -> Any: 

30 if v is None: # for optional fields 

31 return None 

32 

33 # Handle Firestore's DatetimeWithNanoseconds (Read) — normalise to UTC 

34 if hasattr(v, "nanosecond"): 

35 dt = datetime(v.year, v.month, v.day, v.hour, v.minute, v.second, v.microsecond, v.tzinfo) 

36 return dt.astimezone(UTC) if dt.tzinfo else dt.replace(tzinfo=UTC) 

37 # Handle Sentinel objects (Write/Default) 

38 if isinstance(v, Sentinel): 

39 return v 

40 # Handle standard datetime — normalise to UTC 

41 if isinstance(v, datetime): 

42 return v.astimezone(UTC) if v.tzinfo else v.replace(tzinfo=UTC) 

43 # Fallback for strings (if parsing from JSON) 

44 if isinstance(v, str): 

45 try: 

46 dt = datetime.fromisoformat(v) 

47 return dt.astimezone(UTC) if dt.tzinfo else dt.replace(tzinfo=UTC) 

48 except ValueError: 

49 if "Sentinel" in v: 

50 return FirestoreTime.server_timestamp() 

51 return v 

52 

53 

54def serialize_firestore_types(v: Any, info: SerializationInfo) -> Any: 

55 if v is None: # for optional fields 

56 return None 

57 

58 if isinstance(v, Sentinel): 

59 # info.mode is 'json' when calling model_dump_json() 

60 # info.mode is 'python' when calling model_dump() 

61 if info.mode == "json": 

62 return str(v) # Serialize Sentinel as a string for JSON output 

63 return v # Return the actual Sentinel object for Firestore SDK 

64 return v 

65 

66 

67FirestoreField = Annotated[ 

68 datetime | Sentinel, 

69 # This tells Pydantic to just treat it as an 'isinstance' check at runtime 

70 BeforeValidator(validate_firestore_obj), # type: ignore 

71 WithJsonSchema({"type": "string", "format": "date-time"}), 

72 PlainSerializer(serialize_firestore_types, when_used="always"), 

73] 

74 

75OptionalFirestoreField = Annotated[ 

76 datetime | Sentinel | None, 

77 BeforeValidator(validate_firestore_obj), # type: ignore 

78 PlainSerializer(serialize_firestore_types, when_used="always"), 

79]