Coverage for mongodb_odm / utils / utils.py: 98%
60 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-07 08:37 +0000
1import re 1fabcde
2import types 1fabcde
3from typing import Any, Optional, Union 1fabcde
5from mongodb_odm.fields import RelationshipInfo 1fabcde
6from mongodb_odm.utils._internal_models import RelationalFieldInfo 1fabcde
7from pydantic import BaseModel 1fabcde
8from typing_extensions import get_args, get_origin 1fabcde
10UnionType = getattr(types, "UnionType", Union) 1fabcde
11NoneType = type(None) 1fabcde
12pattern = re.compile(r"(?<!^)(?=[A-Z])") 1fabcde
15def camel_to_snake(string: str) -> str: 1fabcde
16 return pattern.sub("_", string).lower() 1fabcde
19def get_database_name(model: Any) -> Optional[Any]: 1fabcde
20 """
21 Get the database name if the model ODMConfig has a user-defined database name.
23 model: Document type
24 """
25 if hasattr(model.ODMConfig, "database"): 1fabcde
26 return model.ODMConfig.database 1fabcde
28 return None 1fabcde
31def convert_model_to_collection(model: Any) -> str: 1fabcde
32 """
33 Get the collection name from the model.
34 Users could define collection names in model ODMConfig.
35 Otherwise, we will convert the model name to a snake case collection name.
37 The user defines collection_name as a higher priority here.
39 model: Document type
40 """
41 if ( 1fab
42 hasattr(model.ODMConfig, "collection_name")
43 and model.ODMConfig.collection_name is not None
44 ):
45 """By default model has ODMConfig in BaseModel""" 1abcde
46 return str(model.ODMConfig.collection_name) 1fabcde
47 else:
48 return camel_to_snake(model.__name__) 1fabcde
51def _is_union_type(t: Any) -> bool: 1fabcde
52 return t is UnionType or t is Union 1fabcde
55def get_type_from_field(field: Any) -> Any: 1fabcde
56 type_: Any = field.annotation 1fabcde
58 # Resolve Optional fields
59 if type_ is None: 1fabcde
60 raise ValueError("Missing field type") 1fabcde
62 origin = get_origin(type_) 1fabcde
64 if origin is None: 1fabcde
65 return type_ 1fabcde
67 if _is_union_type(origin): 1fabcde
68 bases = get_args(type_) 1fabcde
70 if len(bases) > 2: 1fabcde
71 raise ValueError("Cannot have a (non-optional) union as a ODM field") 1fabcde
72 # Non optional unions are not allowed
73 if bases[0] is not NoneType and bases[1] is not NoneType: 1fabcde
74 raise ValueError("Cannot have a (non-optional) union as a ODM field") 1fabcde
76 # Optional unions are allowed
77 return bases[0] if bases[0] is not NoneType else bases[1] 1fabcde
78 elif origin is list: 1fabcde
79 inner_type_ = get_args(type_)[0] 1fabcde
80 return inner_type_ 1fabcde
82 return origin
85def get_model_fields(cls: type[BaseModel]) -> dict[str, Any]: 1fabcde
86 return cls.__pydantic_fields__ 1fabcde
89def _get_fields_info( 1fabcde
90 cls: type[BaseModel], fields: list[str]
91) -> dict[str, RelationalFieldInfo]:
92 """
93 Extract related field information for a model for specific fields.
94 """
95 field_data: dict[str, RelationalFieldInfo] = {} 1fabcde
96 for field in fields: 1fabcde
97 field_type_obj = get_model_fields(cls)[field] 1fabcde
98 if field_type_obj.default.local_field not in get_model_fields(cls): 1fabcde
99 # Check Relationship local_field exists in the model
100 raise Exception( 1fabcde
101 f'Invalid field "{field_type_obj.default.local_field}" in Relationship'
102 )
103 """ 1abcde
104 Example relationship
105 class User(Document):
106 ...
108 class Course(Document):
109 ...
110 author_id: ODMObjectId = Field(...)
111 author: Optional[User] = Relationship(local_field="author_id")
112 """
113 field_type = get_type_from_field(field_type_obj) 1fabcde
114 field_data[field] = RelationalFieldInfo( 1fabcde
115 model=field_type, # User
116 local_field=field_type_obj.default.local_field, # author_id
117 related_field=field_type_obj.default.related_field, # author
118 )
120 return field_data 1fabcde
123def get_relationship_fields_info( 1fabcde
124 cls: type[BaseModel],
125) -> dict[str, RelationalFieldInfo]:
126 """
127 This function will return all of the related fields with their information.
129 cls: Document
130 """
131 fields_name = [] 1fabcde
132 for field_name, field_info in get_model_fields(cls).items(): 1fabcde
133 """Get all fields that are related to a specific model.""" 1abcde
134 if type(field_info.default) is RelationshipInfo: 1fabcde
135 fields_name.append(field_name) 1fabcde
137 return _get_fields_info(cls, fields_name) 1fabcde