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
« 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
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.
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.)
21 Methods:
22 None (The class does not define any methods; it relies on the metaclass
23 and derived classes for functionality.)
24 """
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.
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.
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 """
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
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.
67 Args:
68 data (dict | list): The dictionary or list to process.
70 Returns:
71 dict | list: The processed dictionary or list with modifications.
72 """
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 ]
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
107 def __init__(self, **attrs):
108 self.__replace_empty_dicts(attrs)
109 super().__init__(**attrs)