| 13 | |
| 14 | |
| 15 | def load_str_bytes( |
| 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: |
| 25 | if content_type.endswith(('json', 'javascript')): |
| 26 | pass |
| 27 | elif allow_pickle and content_type.endswith('pickle'): |
| 28 | proto = Protocol.pickle |
| 29 | else: |
| 30 | raise TypeError(f'Unknown content-type: {content_type}') |
| 31 | |
| 32 | proto = proto or Protocol.json |
| 33 | |
| 34 | if proto == Protocol.json: |
| 35 | if isinstance(b, bytes): |
| 36 | b = b.decode(encoding) |
| 37 | return json_loads(b) |
| 38 | elif proto == Protocol.pickle: |
| 39 | if not allow_pickle: |
| 40 | raise RuntimeError('Trying to decode with pickle with allow_pickle=False') |
| 41 | bb = b if isinstance(b, bytes) else b.encode() |
| 42 | return pickle.loads(bb) |
| 43 | else: |
| 44 | raise TypeError(f'Unknown protocol: {proto}') |
| 45 | |
| 46 | |
| 47 | def load_file( |