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

1from typing import Any 

2from bson import ObjectId 

3from pydantic import GetCoreSchemaHandler 

4from pydantic_core import core_schema, CoreSchema 

5from pydantic_core.core_schema import ValidationInfo, str_schema 

6 

7 

8class Id(str): 

9 """ 

10 Extends the base string type to implement custom validation and serialization logic 

11 specifically for IDs that are expected to be in MongoDB's ObjectId format. This class 

12 ensures that IDs are valid ObjectIds and can be used consistently throughout the model 

13 where ID validation and serialization are necessary. 

14 

15 Methods: 

16 validate(cls, v, _): Validates that the given value `v` is a valid ObjectId. If `v` 

17 is None, it returns None. If `v` is invalid, it raises a 

18 ValueError. 

19 serialization(cls, v): Converts a valid ObjectId `v` into a string for serialization. 

20 If `v` is invalid, it raises a ValueError. 

21 __get_pydantic_core_schema__(cls, source_type, handler): Returns a schema for use 

22 with Pydantic, providing metadata about how to handle 

23 the serialization and validation of ObjectIds, ensuring 

24 that data adheres to the expected format both in Python 

25 and JSON. 

26 

27 These methods are class methods, meaning they are called on the class rather than on 

28 instances of the class. 

29 """ 

30 

31 @classmethod 

32 def validate(cls, v, _: ValidationInfo): 

33 if v is None: 

34 return v 

35 if not ObjectId.is_valid(v): 

36 raise ValueError("invalid Id") 

37 return str(v) 

38 

39 # @classmethod 

40 # def serialization(cls, v): 

41 # if not ObjectId.is_valid(v): 

42 # raise ValueError("invalid Id") 

43 # return str(v) 

44 

45 @classmethod 

46 def __get_pydantic_core_schema__( 

47 cls, source_type: Any, handler: GetCoreSchemaHandler 

48 ) -> CoreSchema: # type: ignore 

49 return core_schema.json_or_python_schema( 

50 python_schema=core_schema.with_info_plain_validator_function(cls.validate), 

51 json_schema=str_schema(), 

52 serialization=core_schema.plain_serializer_function_ser_schema( 

53 lambda v: str(v) if ObjectId.is_valid(v) else v 

54 ), 

55 )