Coverage for pydantic/parse.py: 100.00%

1import json 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

2import pickle 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

3from enum import Enum 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

4from pathlib import Path 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

5from typing import Any, Callable, Union 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

6 

7from pydantic.types import StrBytes 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

8 

9 

10class Protocol(str, Enum): 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

11 json = 'json' 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

12 pickle = 'pickle' 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

13 

14 

15def load_str_bytes( 41 ctx1abcdefghijklmnopqrstEFGHPQRSTUIuvwxyzABCD

16 b: StrBytes, 

17 *, 

18 content_type: str = None, 

19 encoding: str = 'utf8', 

20 proto: Protocol = None, 

21 allow_pickle: bool = False, 

22 json_loads: Callable[[str], Any] = json.loads, 

23) -> Any: 

24 if proto is None and content_type: 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

25 if content_type.endswith(('json', 'javascript')): 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

26 pass 36 ctx1JKabcdefghijLMklmnopqrstNOuvwxyzABCD

27 elif allow_pickle and content_type.endswith('pickle'): 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

28 proto = Protocol.pickle 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

29 else: 

30 raise TypeError(f'Unknown content-type: {content_type}') 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

31 

32 proto = proto or Protocol.json 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

33 

34 if proto == Protocol.json: 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

35 if isinstance(b, bytes): 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

36 b = b.decode(encoding) 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

37 return json_loads(b) 47 ctx1JKabcdefghijLMklmnopqrstEFGHPQRSTUINOuvwxyzABCD

38 elif proto == Protocol.pickle: 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

39 if not allow_pickle: 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

40 raise RuntimeError('Trying to decode with pickle with allow_pickle=False') 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

41 bb = b if isinstance(b, bytes) else b.encode() 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

42 return pickle.loads(bb) 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

43 else: 

44 raise TypeError(f'Unknown protocol: {proto}') 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

45 

46 

47def load_file( 41 ctx1abcdefghijklmnopqrstEFGHPQRSTUIuvwxyzABCD

48 path: Union[str, Path], 

49 *, 

50 content_type: str = None, 

51 encoding: str = 'utf8', 

52 proto: Protocol = None, 

53 allow_pickle: bool = False, 

54 json_loads: Callable[[str], Any] = json.loads, 

55) -> Any: 

56 path = Path(path) 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

57 b = path.read_bytes() 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

58 if content_type is None: 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

59 if path.suffix in ('.js', '.json'): 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

60 proto = Protocol.json 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

61 elif path.suffix == '.pkl': 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

62 proto = Protocol.pickle 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

63 

64 return load_str_bytes( 41 ctx1JKabcdefghijLMklmnopqrstEFGHINOuvwxyzABCD

65 b, proto=proto, content_type=content_type, encoding=encoding, allow_pickle=allow_pickle, json_loads=json_loads 

66 )