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

31 statements  

« prev     ^ index     » next       coverage.py v7.7.1, created at 2025-03-27 14:31 +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 if "_id" in data: 

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

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

78 if isinstance(value, dict): 

79 if not value: 

80 data[key] = None 

81 else: 

82 self.__replace_empty_dicts(value) 

83 elif isinstance(value, list): 

84 data[key] = [ 

85 ( 

86 self.__replace_empty_dicts(item) 

87 if isinstance(item, dict) 

88 else item 

89 ) 

90 for item in value 

91 if not (isinstance(item, dict) and not item) 

92 ] 

93 return data 

94 

95 def __init__(self, **attrs): 

96 self.__replace_empty_dicts(attrs) 

97 super().__init__(**attrs)