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
« 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
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 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
95 def __init__(self, **attrs):
96 self.__replace_empty_dicts(attrs)
97 super().__init__(**attrs)