Coverage for pyodmongo/models/db_model.py: 100%

31 statements  

« prev     ^ index     » next       coverage.py v7.6.10, created at 2025-01-16 15:08 +0000

1from pydantic import ConfigDict 

2from .id_model import Id 

3from datetime import datetime 

4from typing import ClassVar 

5from pydantic import BaseModel 

6from typing import ClassVar 

7from .metaclasses import PyOdmongoMeta, DbMeta 

8from pydantic_core import PydanticUndefined 

9 

10 

11class MainBaseModel(BaseModel, metaclass=PyOdmongoMeta): 

12 """ 

13 Base class for all models in PyODMongo, using PyOdmongoMeta as its metaclass. 

14 This class provides the foundational structure for other models, enabling 

15 the integration of database-specific configurations and behaviors. 

16 

17 Attributes: 

18 None (The class itself does not define any attributes; it serves as a 

19 base for other models to extend and utilize the provided metaclass.) 

20 

21 Methods: 

22 None (The class does not define any methods; it relies on the metaclass 

23 and derived classes for functionality.) 

24 """ 

25 

26 

27class DbModel(BaseModel, metaclass=DbMeta): 

28 """ 

29 Base class for all database models using PyODMongo with auto-mapped fields 

30 to MongoDB documents. Provides automatic timestamping and ID management, 

31 along with utilities for managing nested dictionary fields. 

32 

33 Attributes: 

34 id (Id | None): Unique identifier for the database record, typically 

35 mapped to MongoDB's '_id'. 

36 created_at (datetime | None): Timestamp indicating when the record was 

37 created. 

38 updated_at (datetime | None): Timestamp indicating when the record was 

39 last updated. 

40 model_config (ConfigDict): Configuration dictionary to control model 

41 serialization and deserialization behaviors. 

42 _pipeline (ClassVar): Class variable to store pipeline operations for 

43 reference resolution. 

44 

45 Methods: 

46 __init__(**attrs): Initializes a new instance of DbModel, applying 

47 transformations to nested dictionary fields to clean 

48 up empty values. 

49 __remove_empty_dict(dct): Recursively removes empty dictionaries from 

50 nested dictionary fields, aiding in the 

51 cleanup process during initialization. 

52 """ 

53 

54 id: Id | None = None 

55 created_at: datetime | None = None 

56 updated_at: datetime | None = None 

57 model_config = ConfigDict(populate_by_name=True) 

58 _pipeline: ClassVar = [] 

59 _default_language: ClassVar = None 

60 

61 def __replace_empty_dicts(self, data): 

62 """ 

63 Recursively traverses a dictionary (or a list of dictionaries) and: 

64 - Replaces empty dictionaries with None. 

65 - Removes empty dictionaries from lists. 

66 

67 Args: 

68 data (dict | list): The dictionary or list to process. 

69 

70 Returns: 

71 dict | list: The processed dictionary or list with modifications. 

72 """ 

73 

74 if isinstance(data, dict): 

75 # Traverse each key-value pair in the dictionary 

76 if "_id" in data: 

77 data["id"] = data.pop("_id") 

78 for key, value in data.items(): 

79 if isinstance(value, dict): # Check if the value is a dictionary 

80 if not value: # If the dictionary is empty 

81 data[key] = None 

82 else: 

83 data[key] = self.__replace_empty_dicts( 

84 value 

85 ) # Recursive call for non-empty dictionaries 

86 elif isinstance(value, list): # Check if the value is a list 

87 # Process each item in the list and remove empty dictionaries 

88 data[key] = [ 

89 self.__replace_empty_dicts(item) 

90 for item in value 

91 if not ( 

92 isinstance(item, dict) and not item 

93 ) # Exclude empty dictionaries 

94 ] 

95 

96 # elif isinstance(data, list): 

97 # # If the data itself is a list, process each item and remove empty dictionaries 

98 # data = [ 

99 # self.__replace_empty_dicts(item) 

100 # for item in data 

101 # if not ( 

102 # isinstance(item, dict) and not item 

103 # ) # Exclude empty dictionaries 

104 # ] 

105 return data 

106 

107 def __init__(self, **attrs): 

108 self.__replace_empty_dicts(attrs) 

109 super().__init__(**attrs)