|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
3 | | -import base64 |
4 | | -import binascii |
5 | | -from io import BytesIO |
6 | | -from typing import Annotated, Any |
7 | | - |
8 | | -from pydantic import BaseModel, BeforeValidator, ConfigDict, PlainSerializer |
9 | | - |
10 | | - |
11 | | -def _bytesio_from_network(value: Any) -> BytesIO: |
12 | | - if isinstance(value, BytesIO): |
13 | | - return value |
14 | | - if isinstance(value, (bytes, bytearray)): |
15 | | - return BytesIO(bytes(value)) |
16 | | - if isinstance(value, str): |
17 | | - try: |
18 | | - raw = base64.b64decode(value.encode("ascii"), validate=True) |
19 | | - except (ValueError, UnicodeEncodeError, binascii.Error) as exc: |
20 | | - raise ValueError("Invalid base64 string for BytesIO.") from exc |
21 | | - return BytesIO(raw) |
22 | | - raise TypeError(f"Expected BytesIO/base64 string/bytes, got: {type(value)!r}") |
23 | | - |
24 | | - |
25 | | -def _bytesio_to_network(value: BytesIO) -> str: |
26 | | - return base64.b64encode(value.getvalue()).decode("ascii") |
27 | | - |
28 | | - |
29 | | -Base64BytesIO = Annotated[ |
30 | | - BytesIO, |
31 | | - BeforeValidator(_bytesio_from_network), |
32 | | - PlainSerializer(_bytesio_to_network, return_type=str, when_used="json"), |
33 | | -] |
| 3 | +from pydantic import BaseModel, ConfigDict |
34 | 4 |
|
35 | 5 |
|
36 | 6 | class NetworkModel(BaseModel): |
|
0 commit comments