Coverage for src/prisma/validator.py: 100%

1import sys 

2from types import ModuleType 

3from typing import Any, Type, TypeVar, cast 

4from functools import lru_cache 

5 

6from pydantic import BaseModel 

7 

8from ._types import Protocol, runtime_checkable 

9from ._compat import PYDANTIC_V2, Extra, is_typeddict 

10 

11__all__ = ('validate',) 

12 

13# NOTE: we should use bound=TypedDict but mypy does not support this 

14T = TypeVar('T') 

15 

16 

17class Config: 

18 extra: Extra = Extra.forbid 

19 

20 

21@runtime_checkable 

22class CachedModel(Protocol): 

23 __pydantic_model__: BaseModel 

24 

25 

26def _get_module(typ: Type[Any]) -> ModuleType: 

27 return sys.modules[typ.__module__] 

28 

29 

30@lru_cache(maxsize=None) 

31def patch_pydantic() -> None: 

32 """Pydantic does not resolve forward references for TypedDict types properly yet 

33 

34 see https://github.com/samuelcolvin/pydantic/pull/2761 

35 """ 

36 

37 annotated_types: Any 

38 if PYDANTIC_V2: 

39 from pydantic.v1 import annotated_types 

40 else: 

41 from pydantic import annotated_types # type: ignore[no-redef] 

42 

43 create_model = annotated_types.create_model_from_typeddict 

44 

45 def patched_create_model(typeddict_cls: Any, **kwargs: Any) -> Type[BaseModel]: 

46 kwargs.setdefault('__module__', typeddict_cls.__module__) 

47 return create_model(typeddict_cls, **kwargs) # type: ignore[no-any-return] 

48 

49 annotated_types.create_model_from_typeddict = patched_create_model 

50 

51 

52# Note: we can't just use TypeAdapter in v2 due to this issue 

53# https://github.com/pydantic/pydantic/issues/7111 

54 

55 

56def validate(type: Type[T], data: Any) -> T: 

57 """Validate untrusted data matches a given TypedDict 

58 

59 For example: 

60 

61 from prisma import validate, types 

62 from prisma.models import User 

63 

64 def user_create_handler(data: Any) -> None: 

65 validated = validate(types.UserCreateInput, data) 

66 user = await User.prisma().create(data=validated) 

67 """ 

68 create_model_from_typeddict: Any 

69 if PYDANTIC_V2: 

70 from pydantic.v1 import create_model_from_typeddict 

71 else: 

72 from pydantic import create_model_from_typeddict # type: ignore 

73 

74 # avoid patching pydantic until we know we need to in case our 

75 # monkey patching fails 

76 patch_pydantic() 

77 

78 if not is_typeddict(type): 

79 raise TypeError(f'Only TypedDict types are supported, got: {type} instead.') 

80 

81 # we cannot use pydantic's builtin type -> model resolver 

82 # as we need to be able to update forward references 

83 if isinstance(type, CachedModel): 

84 # cache the model on the type object, mirroring how pydantic works 

85 # mypy thinks this is unreachable, we know it isn't, just ignore 

86 model = type.__pydantic_model__ # type: ignore[unreachable] 

87 else: 

88 # pyright is more strict than mypy here, we also don't care about the 

89 # incorrectly inferred type as we have verified that the given type 

90 # is indeed a TypedDict 

91 model = create_model_from_typeddict( 

92 type, 

93 __config__=Config, # pyright: ignore[reportGeneralTypeIssues] 

94 ) 

95 model.update_forward_refs(**vars(_get_module(type))) 

96 type.__pydantic_model__ = model # type: ignore 

97 

98 instance = model.parse_obj(data) 

99 return cast(T, instance.dict(exclude_unset=True))